remove synchronized

wooEnrico · Jun 25, 2024 · 222906b · 222906b
1 parent d921a43
commit 222906b
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 9 deletions.
diff --git a/kafka/src/main/java/io/github/wooenrico/kafka/consumer/KafkaConsumer.java b/kafka/src/main/java/io/github/wooenrico/kafka/consumer/KafkaConsumer.java
@@ -71,19 +71,14 @@ public void close() {
         this.threadPoolExecutor.shutdown();
     }
 
-    private synchronized void subscribe(org.apache.kafka.clients.consumer.KafkaConsumer<K, V> kafkaConsumer) {
+    private void subscribe(org.apache.kafka.clients.consumer.KafkaConsumer<K, V> kafkaConsumer) {
         if (kafkaConsumer != null) {
             CompletableFuture<Void> remove = this.consumerMap.remove(kafkaConsumer);
-            // kafka close is not thread-safe must call by poll thread
-            CompletableFuture.runAsync(kafkaConsumer::close, this.pollExecutor);
             if (remove == null) {
                 return;
             }
-
-            try {
-                Thread.sleep(0);
-            } catch (InterruptedException ignored) {
-            }
+            // kafka close is not thread-safe must call by poll thread
+            CompletableFuture.runAsync(kafkaConsumer::close, this.pollExecutor);
         }
 
         if (this.close.get()) {

diff --git a/kafka/src/main/java/io/github/wooenrico/kafka/consumer/ReactorKafkaReceiver.java b/kafka/src/main/java/io/github/wooenrico/kafka/consumer/ReactorKafkaReceiver.java
@@ -65,7 +65,7 @@ public void close() throws IOException {
         this.threadPoolExecutor.shutdown();
     }
 
-    private synchronized void subscribe(KafkaReceiver<K, V> kafkaReceiver) {
+    private void subscribe(KafkaReceiver<K, V> kafkaReceiver) {
         if (kafkaReceiver != null) {
             Disposable remove = this.subscribers.remove(kafkaReceiver);
             if (remove == null) {