EtcdDataSource#initWatcher make "Thread blocked"

[LProDreamAll]

EtcdDataSource#initWatcher make "Thread blocked"

The EtcdDataSource class reported Thread blocked When using etcd to manage rules

1. WARN 49121 --- [-thread-checker] io.vertx.core.impl.BlockedThreadChecker : Thread Thread[vert.x-eventloop-thread-0,5,main] has been blocked for 5225 ms, time limit is 2000 ms.
   io.vertx.core.VertxException: Thread blocked
   at sun.misc.Unsafe.park(Native Method) ~[na:1.8.0_221]
   at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) ~[na:1.8.0_221]
   at java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1693) ~[na:1.8.0_221]
   at java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3323) ~[na:1.8.0_221]
   at java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1729) ~[na:1.8.0_221]
   at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895) ~[na:1.8.0_221]
   at com.alibaba.csp.sentinel.datasource.etcd.EtcdDataSource.readSource(EtcdDataSource.java:114) ~[classes/:na]

com.alibaba.csp.sentinel.datasource.etcd.EtcdDataSource#initWatcher

private void initWatcher() {
watcher = client.getWatchClient().watch(ByteSequence.from(key, charset), (watchResponse) -> {
for (WatchEvent watchEvent : watchResponse.getEvents()) {
WatchEvent.EventType eventType = watchEvent.getEventType();
if (eventType == WatchEvent.EventType.PUT) {
try {
T newValue = loadConfig();
getProperty().updateValue(newValue);
} catch (Exception e) {
RecordLog.warn("[EtcdDataSource] Failed to update config", e);
}
} else if (eventType == WatchEvent.EventType.DELETE) {
RecordLog.info("[EtcdDataSource] Cleaning config for key <{}>", key);
getProperty().updateValue(null);
}
}
});
}

Solution

private void initWatcher() {
watcher = client.getWatchClient().watch(ByteSequence.from(key, charset), (watchResponse) -> {
for (WatchEvent watchEvent : watchResponse.getEvents()) {
WatchEvent.EventType eventType = watchEvent.getEventType();
final String nodeValue = watchEvent.getKeyValue().getValue().toString(StandardCharsets.UTF_8);
if (eventType == WatchEvent.EventType.PUT) {
try {
T newValue = parser.convert(nodeValue);
getProperty().updateValue(newValue);
} catch (Exception e) {
RecordLog.warn("[EtcdDataSource] Failed to update config", e);
}
} else if (eventType == WatchEvent.EventType.DELETE) {
RecordLog.info("[EtcdDataSource] Cleaning config for key <{}>", key);
getProperty().updateValue(null);
}
}
});
}

[sczyh30]

Please refer #2991

[LProDreamAll]

@sczyh30 I close that