KAFKA-13971:Atomicity violations caused by improper usage of ConcurrentHashMap
Reviewers: Divij Vaidya Chris Egerton
@@ -78,25 +78,20 @@ public class WorkerConfigTransformer implements AutoCloseable {
}
private void scheduleReload(String connectorName, String path, long ttl) {
- Map<String, HerderRequest> connectorRequests = requests.get(connectorName);
- if (connectorRequests == null) {
- connectorRequests = new ConcurrentHashMap<>();
- requests.put(connectorName, connectorRequests);
- } else {
- HerderRequest previousRequest = connectorRequests.get(path);
+ Map<String, HerderRequest> connectorRequests = requests.computeIfAbsent(connectorName, s -> new ConcurrentHashMap<>());
+ connectorRequests.compute(path, (s, previousRequest) -> {
if (previousRequest != null) {
// Delete previous request for ttl which is now stale
previousRequest.cancel();
}
- }
log.info("Scheduling a restart of connector {} in {} ms", connectorName, ttl);
Callback<Void> cb = (error, result) -> {
if (error != null) {
log.error("Unexpected error during connector restart: ", error);
}
};
- HerderRequest request = worker.herder().restartConnector(ttl, connectorName, cb);
- connectorRequests.put(path, request);
+ return worker.herder().restartConnector(ttl, connectorName, cb);
+ });
}
@Override
