opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-29 04:13:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE 4926]Fix FileWatchServiceTest (#4927)

Browse Source 
* Fix FileWatchService

* Fix all warnings and optimize FileWatchService implementation

* Take spurious wakeup into account
This commit is contained in:
Zhanhui Li
2022-08-30 14:24:46 +08:00
committed by GitHub
parent 7f25e453cd
commit a488a0c22d
5 changed files with 156 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/src/main/java/org/apache/rocketmq/common/LifecycleAwareServiceThread.java
+56
View File
@@ -0,0 +1,56 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.common;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public abstract class LifecycleAwareServiceThread extends ServiceThread {
private final AtomicBoolean started = new AtomicBoolean(false);
@Override
public void run() {
started.set(true);
synchronized (started) {
started.notifyAll();
}
run0();
}
public abstract void run0();
/**
* Take spurious wakeup into account.
*
* @param timeout amount of time in milliseconds
* @throws InterruptedException if interrupted
*/
public void awaitStarted(long timeout) throws InterruptedException {
long expire = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeout);
synchronized (started) {
while (!started.get()) {
long duration = expire - System.nanoTime();
if (duration < TimeUnit.MILLISECONDS.toNanos(1)) {
break;
}
started.wait(TimeUnit.NANOSECONDS.toMillis(duration));
}
}
}
}
srvutil/pom.xml
+5
View File
@@ -49,5 +49,10 @@
<groupId>com.googlecode.concurrentlinkedhashmap</groupId>
<artifactId>concurrentlinkedhashmap-lru</artifactId>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>
srvutil/src/main/java/org/apache/rocketmq/srvutil/FileWatchService.java
+42 -30
View File
@@ -24,33 +24,28 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.List;
import org.apache.rocketmq.common.ServiceThread;
import java.util.HashMap;
import java.util.Map;
import org.apache.rocketmq.common.LifecycleAwareServiceThread;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
public class FileWatchService extends ServiceThread {
public class FileWatchService extends LifecycleAwareServiceThread {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
private final List<String> watchFiles;
private final List<String> fileCurrentHash;
private final Map<String, String> currentHash = new HashMap<>();
private final Listener listener;
private static final int WATCH_INTERVAL = 500;
private MessageDigest md = MessageDigest.getInstance("MD5");
private final MessageDigest md = MessageDigest.getInstance("MD5");
public FileWatchService(final String[] watchFiles,
final Listener listener) throws Exception {
this.listener = listener;
this.watchFiles = new ArrayList<>();
this.fileCurrentHash = new ArrayList<>();
for (int i = 0; i < watchFiles.length; i++) {
if (!Strings.isNullOrEmpty(watchFiles[i]) && new File(watchFiles[i]).exists()) {
this.watchFiles.add(watchFiles[i]);
this.fileCurrentHash.add(hash(watchFiles[i]));
for (String file : watchFiles) {
if (!Strings.isNullOrEmpty(file) && new File(file).exists()) {
currentHash.put(file, md5Digest(file));
}
}
}
@@ -61,36 +56,52 @@ public class FileWatchService extends ServiceThread {
}
@Override
public void run() {
public void run0() {
log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
try {
this.waitForRunning(WATCH_INTERVAL);
for (int i = 0; i < watchFiles.size(); i++) {
String newHash;
try {
newHash = hash(watchFiles.get(i));
} catch (Exception ignored) {
log.warn(this.getServiceName() + " service has exception when calculate the file hash. ", ignored);
continue;
}
if (!newHash.equals(fileCurrentHash.get(i))) {
fileCurrentHash.set(i, newHash);
listener.onChanged(watchFiles.get(i));
for (Map.Entry<String, String> entry : currentHash.entrySet()) {
String newHash = md5Digest(entry.getKey());
if (!newHash.equals(currentHash.get(entry.getKey()))) {
entry.setValue(newHash);
listener.onChanged(entry.getKey());
}
}
} catch (Exception e) {
log.warn(this.getServiceName() + " service has exception. ", e);
log.warn(this.getServiceName() + " service raised an unexpected exception.", e);
}
}
log.info(this.getServiceName() + " service end");
}
private String hash(String filePath) throws IOException {
/**
* Note: we ignore DELETE event on purpose. This is useful when application renew CA file.
* When the operator delete/rename the old CA file and copy a new one, this ensures the old CA file is used during
* the operation.
* <p>
* As we know exactly what to do when file does not exist or when IO exception is raised, there is no need to
* propagate the exception up.
*
* @param filePath Absolute path of the file to calculate its MD5 digest.
* @return Hash of the file content if exists; empty string otherwise.
*/
private String md5Digest(String filePath) {
Path path = Paths.get(filePath);
md.update(Files.readAllBytes(path));
if (!path.toFile().exists()) {
// Reuse previous hash result
return currentHash.getOrDefault(filePath, "");
}
byte[] raw;
try {
raw = Files.readAllBytes(path);
} catch (IOException e) {
log.info("Failed to read content of {}", filePath);
// Reuse previous hash result
return currentHash.getOrDefault(filePath, "");
}
md.update(raw);
byte[] hash = md.digest();
return UtilAll.bytes2string(hash);
}
@@ -98,6 +109,7 @@ public class FileWatchService extends ServiceThread {
public interface Listener {
/**
* Will be called when the target files are changed
*
* @param path the changed file path
*/
void onChanged(String path);
srvutil/src/test/java/logback-test.xml
+28
View File
@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<root level="debug">
<appender-ref ref="STDOUT" />
</root>
</configuration>
srvutil/src/test/java/org/apache/rocketmq/srvutil/FileWatchServiceTest.java
+25 -33
View File
@@ -20,6 +20,8 @@ package org.apache.rocketmq.srvutil;
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.junit.Rule;
@@ -39,17 +41,16 @@ public class FileWatchServiceTest {
public void watchSingleFile() throws Exception {
final File file = tempFolder.newFile();
final Semaphore waitSemaphore = new Semaphore(0);
FileWatchService fileWatchService = new FileWatchService(new String[] {file.getAbsolutePath()}, new FileWatchService.Listener() {
@Override
public void onChanged(String path) {
assertThat(file.getAbsolutePath()).isEqualTo(path);
waitSemaphore.release();
}
FileWatchService fileWatchService = new FileWatchService(new String[] {file.getAbsolutePath()}, path -> {
assertThat(file.getAbsolutePath()).isEqualTo(path);
waitSemaphore.release();
});
fileWatchService.start();
fileWatchService.awaitStarted(1000);
modifyFile(file);
boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isTrue();
fileWatchService.shutdown();
}
@Test
@@ -57,46 +58,42 @@ public class FileWatchServiceTest {
File file = tempFolder.newFile();
final Semaphore waitSemaphore = new Semaphore(0);
FileWatchService fileWatchService = new FileWatchService(new String[] {file.getAbsolutePath()},
new FileWatchService.Listener() {
@Override
public void onChanged(String path) {
waitSemaphore.release();
}
});
path -> waitSemaphore.release());
fileWatchService.start();
file.delete();
fileWatchService.awaitStarted(1000);
assertThat(file.delete()).isTrue();
boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isFalse();
file.createNewFile();
assertThat(file.createNewFile()).isTrue();
modifyFile(file);
result = waitSemaphore.tryAcquire(1, 2000, TimeUnit.MILLISECONDS);
assertThat(result).isTrue();
fileWatchService.shutdown();
}
@Test
public void watchTwoFile_FileDeleted() throws Exception {
File fileA = tempFolder.newFile();
File fileB = tempFolder.newFile();
Files.write(fileA.toPath(), "Hello, World!".getBytes(StandardCharsets.UTF_8));
Files.write(fileB.toPath(), "Hello, World!".getBytes(StandardCharsets.UTF_8));
final Semaphore waitSemaphore = new Semaphore(0);
FileWatchService fileWatchService = new FileWatchService(
new String[] {fileA.getAbsolutePath(), fileB.getAbsolutePath()},
new FileWatchService.Listener() {
@Override
public void onChanged(String path) {
waitSemaphore.release();
}
});
path -> waitSemaphore.release());
fileWatchService.start();
fileA.delete();
fileWatchService.awaitStarted(1000);
assertThat(fileA.delete()).isTrue();
boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isFalse();
modifyFile(fileB);
result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isTrue();
fileA.createNewFile();
assertThat(fileA.createNewFile()).isTrue();
modifyFile(fileA);
result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isTrue();
fileWatchService.shutdown();
}
@Test
@@ -106,17 +103,16 @@ public class FileWatchServiceTest {
final Semaphore waitSemaphore = new Semaphore(0);
FileWatchService fileWatchService = new FileWatchService(
new String[] {fileA.getAbsolutePath(), fileB.getAbsolutePath()},
new FileWatchService.Listener() {
@Override
public void onChanged(String path) {
path -> {
assertThat(path).isEqualTo(fileA.getAbsolutePath());
waitSemaphore.release();
}
});
});
fileWatchService.start();
fileWatchService.awaitStarted(1000);
modifyFile(fileA);
boolean result = waitSemaphore.tryAcquire(1, 1000, TimeUnit.MILLISECONDS);
assertThat(result).isTrue();
fileWatchService.shutdown();
}
@Test
@@ -126,13 +122,9 @@ public class FileWatchServiceTest {
final Semaphore waitSemaphore = new Semaphore(0);
FileWatchService fileWatchService = new FileWatchService(
new String[] {fileA.getAbsolutePath(), fileB.getAbsolutePath()},
new FileWatchService.Listener() {
@Override
public void onChanged(String path) {
waitSemaphore.release();
}
});
path -> waitSemaphore.release());
fileWatchService.start();
fileWatchService.awaitStarted(1000);
modifyFile(fileA);
modifyFile(fileB);
boolean result = waitSemaphore.tryAcquire(2, 1000, TimeUnit.MILLISECONDS);