[test] use AtomicBoolean

apache · Jan 12, 2024 · 53447ff · 53447ff
1 parent 65e4c74
commit 53447ff
Showing 1 changed file with 6 additions and 5 deletions.
diff --git a/...java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java b/...java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
@@ -64,6 +64,7 @@
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 /**
@@ -93,7 +94,7 @@ public class DynamicKafkaSourceReader<T> implements SourceReader<T, DynamicKafka
     private MultipleFuturesAvailabilityHelper availabilityHelper;
     private boolean isActivelyConsumingSplits;
     private boolean isNoMoreSplits;
-    private boolean restartingReaders;
+    private AtomicBoolean restartingReaders;
 
     public DynamicKafkaSourceReader(
             SourceReaderContext readerContext,
@@ -113,7 +114,7 @@ public DynamicKafkaSourceReader(
         this.availabilityHelper = new MultipleFuturesAvailabilityHelper(0);
         this.isNoMoreSplits = false;
         this.isActivelyConsumingSplits = false;
-        this.restartingReaders = false;
+        this.restartingReaders = new AtomicBoolean();
         this.clustersProperties = new HashMap<>();
     }
 
@@ -136,7 +137,7 @@ public InputStatus pollNext(ReaderOutput<T> readerOutput) throws Exception {
             return logAndReturnInputStatus(InputStatus.NOTHING_AVAILABLE);
         }
 
-        if (restartingReaders) {
+        if (restartingReaders.get()) {
             logger.info("Poll next invoked while restarting readers");
             return logAndReturnInputStatus(InputStatus.NOTHING_AVAILABLE);
         }
@@ -282,7 +283,7 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
         // comes at the cost of complexity and it is an optimization for a corner case. We can
         // revisit if necessary.
         if (!newClustersAndTopics.equals(currentMetadataFromState)) {
-            restartingReaders = true;
+            restartingReaders.set(true);
             closeAllReadersAndClearState();
 
             clustersProperties.putAll(newClustersProperties);
@@ -485,7 +486,7 @@ private void completeAndResetAvailabilityHelper() {
                 .getAvailableFuture()
                 .whenComplete(
                         (ignore, t) -> {
-                            restartingReaders = false;
+                            restartingReaders.set(false);
                             cachedPreviousFuture.complete(null);
                         });
     }