(chore) renames

leancloud · Mar 9, 2020 · f60d1fe · f60d1fe
1 parent 37bc1d7
commit f60d1fe
Show file tree

Hide file tree

Showing 10 changed files with 35 additions and 30 deletions.
diff --git a/src/main/java/cn/leancloud/kafka/consumer/AbstractCommitPolicy.java b/src/main/java/cn/leancloud/kafka/consumer/AbstractCommitPolicy.java
@@ -8,6 +8,8 @@
 import java.time.Duration;
 import java.util.*;
 
+import static java.util.Collections.emptySet;
+
 abstract class AbstractCommitPolicy<K, V> implements CommitPolicy {
     static SleepFunction sleepFunction = Thread::sleep;
 
@@ -55,19 +57,19 @@ void onError(RetriableException e) {
     public Set<TopicPartition> partialCommitSync(ProcessRecordsProgress progress) {
         final Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = progress.completedOffsetsToCommit();
         if (offsetsToCommit.isEmpty()) {
-            return Collections.emptySet();
+            return emptySet();
         }
         commitSyncWithRetry(offsetsToCommit);
         progress.updateCommittedOffsets(offsetsToCommit);
 
-        return progress.clearProgressForCompletedPartitions(offsetsToCommit);
+        return progress.clearCompletedPartitions(offsetsToCommit);
     }
 
     Set<TopicPartition> fullCommitSync(ProcessRecordsProgress progress) {
         commitSyncWithRetry();
 
-        final Set<TopicPartition> completePartitions = progress.partitionsForAllPendingRecords();
-        progress.clearAllProgress();
+        final Set<TopicPartition> completePartitions = progress.allPartitions();
+        progress.clearAll();
         return completePartitions;
     }
 

diff --git a/src/main/java/cn/leancloud/kafka/consumer/AsyncCommitPolicy.java b/src/main/java/cn/leancloud/kafka/consumer/AsyncCommitPolicy.java
@@ -34,7 +34,7 @@ Set<TopicPartition> tryCommit0(boolean noPendingRecords, ProcessRecordsProgress
             return emptySet();
         }
 
-        final Set<TopicPartition> partitions = progress.partitionsForAllPendingRecords();
+        final Set<TopicPartition> partitions = progress.allPartitions();
         commit(progress);
 
         // for our commit policy, no matter syncCommit or asyncCommit we are using, we always
@@ -60,7 +60,7 @@ private void commit(ProcessRecordsProgress progress) {
             commitSyncWithRetry();
             pendingAsyncCommitCounter = 0;
             forceSync = false;
-            progress.clearAllProgress();
+            progress.clearAll();
         } else {
             ++pendingAsyncCommitCounter;
             consumer.commitAsync(((offsets, exception) -> {
@@ -70,7 +70,7 @@ private void commit(ProcessRecordsProgress progress) {
                     logger.warn("Failed to commit offsets: " + offsets + " asynchronously", exception);
                     forceSync = true;
                 } else {
-                    progress.clearProgressForCompletedPartitions(offsets);
+                    progress.clearCompletedPartitions(offsets);
                 }
             }));
         }

diff --git a/src/main/java/cn/leancloud/kafka/consumer/AutoCommitPolicy.java b/src/main/java/cn/leancloud/kafka/consumer/AutoCommitPolicy.java
@@ -22,11 +22,11 @@ public Set<TopicPartition> tryCommit(boolean noPendingRecords, ProcessRecordsPro
 
         final Set<TopicPartition> partitions;
         if (noPendingRecords) {
-            partitions = progress.partitionsForAllPendingRecords();
-            progress.clearAllProgress();
+            partitions = progress.allPartitions();
+            progress.clearAll();
         } else {
             partitions = progress.completedPartitions();
-            progress.clearProgressFor(partitions);
+            progress.clearFor(partitions);
         }
 
         return partitions;

diff --git a/src/main/java/cn/leancloud/kafka/consumer/NoOpCommitPolicy.java b/src/main/java/cn/leancloud/kafka/consumer/NoOpCommitPolicy.java
@@ -2,9 +2,10 @@
 
 import org.apache.kafka.common.TopicPartition;
 
-import java.util.Collections;
 import java.util.Set;
 
+import static java.util.Collections.emptySet;
+
 final class NoOpCommitPolicy implements CommitPolicy {
     private static final NoOpCommitPolicy INSTANCE = new NoOpCommitPolicy();
 
@@ -14,11 +15,11 @@ static NoOpCommitPolicy getInstance() {
 
     @Override
     public Set<TopicPartition> tryCommit(boolean noPendingRecords, ProcessRecordsProgress progress) {
-        return Collections.emptySet();
+        return emptySet();
     }
 
     @Override
     public Set<TopicPartition> partialCommitSync(ProcessRecordsProgress progress) {
-        return Collections.emptySet();
+        return emptySet();
     }
 }
diff --git a/src/main/java/cn/leancloud/kafka/consumer/PartialAsyncCommitPolicy.java b/src/main/java/cn/leancloud/kafka/consumer/PartialAsyncCommitPolicy.java
@@ -72,7 +72,7 @@ Set<TopicPartition> tryCommit0(boolean noPendingRecords, ProcessRecordsProgress
                     forceSync = true;
                 } else {
                     progress.updateCommittedOffsets(committedOffsets);
-                    progress.clearProgressForCompletedPartitions(committedOffsets);
+                    progress.clearCompletedPartitions(committedOffsets);
                 }
             });
         }
@@ -123,7 +123,7 @@ private boolean useSyncCommit() {
     }
 
     private Set<TopicPartition> fullCommit(ProcessRecordsProgress progress) {
-        final Set<TopicPartition> completePartitions = progress.partitionsForAllPendingRecords();
+        final Set<TopicPartition> completePartitions = progress.allPartitions();
         if (useSyncCommit()) {
             commitSyncWithRetry();
             pendingAsyncCommitCounter = 0;
@@ -134,11 +134,11 @@ private Set<TopicPartition> fullCommit(ProcessRecordsProgress progress) {
                 assert pendingAsyncCommitCounter >= 0 : "actual: " + pendingAsyncCommitCounter;
                 if (exception != null) {
                     // if last async commit is failed, we do not clean cached completed offsets and let next
-                    // commit be a sync commit so all the complete offsets will be committed at that time
+                    // commit be a sync commit so all the completed offsets will be committed at that time
                     logger.warn("Failed to commit offset: " + offsets + " asynchronously", exception);
                     forceSync = true;
                 } else {
-                    progress.clearProgressForCompletedPartitions(offsets);
+                    progress.clearCompletedPartitions(offsets);
                 }
             });
         }
@@ -149,7 +149,7 @@ private Set<TopicPartition> fullCommit(ProcessRecordsProgress progress) {
 
         // we can clear records states even though the async commit may fail. because if
         // it did failed, we will do a sync commit on next try commit
-        progress.clearAllProgress();
+        progress.clearAll();
         pendingAsyncCommitOffset.clear();
         return completePartitions;
     }

diff --git a/src/main/java/cn/leancloud/kafka/consumer/ProcessRecordsProgress.java b/src/main/java/cn/leancloud/kafka/consumer/ProcessRecordsProgress.java
@@ -6,6 +6,7 @@
 
 import java.util.*;
 
+import static java.util.Collections.emptyMap;
 import static java.util.stream.Collectors.toSet;
 
 class ProcessRecordsProgress {
@@ -53,12 +54,12 @@ void markCompletedRecord(ConsumerRecord<?, ?> record) {
         }
     }
 
-    void clearAllProgress() {
+    void clearAll() {
         topicOffsetHighWaterMark.clear();
         completedOffsets.clear();
     }
 
-    void clearProgressFor(Collection<TopicPartition> partitions) {
+    void clearFor(Collection<TopicPartition> partitions) {
         for (TopicPartition p : partitions) {
             topicOffsetHighWaterMark.remove(p);
             completedOffsets.remove(p);
@@ -74,7 +75,7 @@ boolean noPendingRecords() {
         return topicOffsetHighWaterMark.isEmpty();
     }
 
-    Set<TopicPartition> partitionsForAllPendingRecords() {
+    Set<TopicPartition> allPartitions() {
         return new HashSet<>(topicOffsetHighWaterMark.keySet());
     }
 
@@ -84,7 +85,7 @@ boolean noCompletedRecords() {
 
     Map<TopicPartition, OffsetAndMetadata> completedOffsetsToCommit() {
         if (noCompletedRecords()) {
-            return Collections.emptyMap();
+            return emptyMap();
         }
 
         final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
@@ -120,9 +121,9 @@ void updateCommittedOffsets(Map<TopicPartition, OffsetAndMetadata> offsets) {
         }
     }
 
-    Set<TopicPartition> clearProgressForCompletedPartitions(Map<TopicPartition, OffsetAndMetadata> committedOffsets) {
+    Set<TopicPartition> clearCompletedPartitions(Map<TopicPartition, OffsetAndMetadata> committedOffsets) {
         final Set<TopicPartition> partitions = completedPartitions(committedOffsets);
-        clearProgressFor(partitions);
+        clearFor(partitions);
         return partitions;
     }
 

diff --git a/src/main/java/cn/leancloud/kafka/consumer/RebalanceListener.java b/src/main/java/cn/leancloud/kafka/consumer/RebalanceListener.java
@@ -98,7 +98,7 @@ private void pausePreviousPausedPartitions(Collection<TopicPartition> partitions
     private void clearProgressForRevokedPartitions(Collection<TopicPartition> revokedPartitions) {
         // revoke those partitions which was revoked and not reassigned
         if (!revokedPartitions.isEmpty()) {
-            progress.clearProgressFor(revokedPartitions);
+            progress.clearFor(revokedPartitions);
         }
     }
 }
diff --git a/src/main/java/cn/leancloud/kafka/consumer/SyncCommitPolicy.java b/src/main/java/cn/leancloud/kafka/consumer/SyncCommitPolicy.java
@@ -4,9 +4,10 @@
 import org.apache.kafka.common.TopicPartition;
 
 import java.time.Duration;
-import java.util.Collections;
 import java.util.Set;
 
+import static java.util.Collections.emptySet;
+
 final class SyncCommitPolicy<K, V> extends AbstractRecommitAwareCommitPolicy<K, V> {
     SyncCommitPolicy(Consumer<K, V> consumer,
                      Duration syncCommitRetryInterval,
@@ -18,7 +19,7 @@ final class SyncCommitPolicy<K, V> extends AbstractRecommitAwareCommitPolicy<K,
     @Override
     Set<TopicPartition> tryCommit0(boolean noPendingRecords, ProcessRecordsProgress progress) {
         if (!noPendingRecords || progress.noOffsetsToCommit()) {
-            return Collections.emptySet();
+            return emptySet();
         }
 
         final Set<TopicPartition> completePartitions = fullCommitSync(progress);

diff --git a/src/test/java/cn/leancloud/kafka/consumer/ProcessRecordsProgressTest.java b/src/test/java/cn/leancloud/kafka/consumer/ProcessRecordsProgressTest.java
@@ -98,7 +98,7 @@ public void testRevokePartitions() {
             progress.markCompletedRecord(record);
         }
 
-        progress.clearProgressFor(toPartitions(IntStream.range(101, 103).boxed().collect(toList())));
+        progress.clearFor(toPartitions(IntStream.range(101, 103).boxed().collect(toList())));
 
         assertThat(progress.completedOffsetsToCommit())
                 .hasSize(1)

diff --git a/src/test/java/cn/leancloud/kafka/consumer/RebalanceListenerTest.java b/src/test/java/cn/leancloud/kafka/consumer/RebalanceListenerTest.java
@@ -64,12 +64,12 @@ public void testRevokePartitions() {
         doAnswer(invocation -> {
             assertThat(revokedPartitions).containsExactlyInAnyOrderElementsOf(invocation.getArgument(0));
             return null;
-        }).when(progress).clearProgressFor(anySet());
+        }).when(progress).clearFor(anySet());
 
         listener.onPartitionsRevoked(allPartitions);
         listener.onPartitionsAssigned(reAssignedPartitions);
 
-        verify(progress, times(1)).clearProgressFor(anyCollection());
+        verify(progress, times(1)).clearFor(anyCollection());
     }
 
     @Test