Rename caller protected methods to use fetch prefix

linkedin · Oct 6, 2023 · 6d3d2d1 · 6d3d2d1
1 parent 6223bc1
commit 6d3d2d1
Showing 1 changed file with 13 additions and 7 deletions.
diff --git a/...n/src/main/java/com/linkedin/venice/kafka/partitionoffset/PartitionOffsetFetcherImpl.java b/...n/src/main/java/com/linkedin/venice/kafka/partitionoffset/PartitionOffsetFetcherImpl.java
@@ -158,19 +158,22 @@ private long getEarliestOrLatestOffset(
   public long getPartitionOffsetByTime(PubSubTopicPartition pubSubTopicPartition, long timestamp) {
     isTopicPartitionValid(pubSubTopicPartition);
     try (AutoCloseableLock ignore = AutoCloseableLock.of(pubSubConsumerLock)) {
-      Long result = offsetsForTimesUnprotected(pubSubTopicPartition, timestamp);
+      Long result = fetchOffsetsForTimes(pubSubTopicPartition, timestamp);
       if (result == null) {
-        result = getOffsetByTimeIfOutOfRangeUnprotected(pubSubTopicPartition, timestamp);
+        result = fetchOffsetByTimeIfOutOfRange(pubSubTopicPartition, timestamp);
       } else if (result == -1L) {
         // The given timestamp exceed the timestamp of the last message. So return the last offset.
         logger.warn("Offsets result is empty. Will complement with the last offsets.");
-        result = endOffsetsWithRetryUnprotected(pubSubTopicPartition) + 1;
+        result = fetchEndOffsetsWithRetry(pubSubTopicPartition) + 1;
       }
       return result;
     }
   }
 
-  private Long offsetsForTimesUnprotected(PubSubTopicPartition pubSubTopicPartition, long timestamp) {
+  /**
+   * This method needs to be called with the pubSubConsumerLock held.
+   */
+  private Long fetchOffsetsForTimes(PubSubTopicPartition pubSubTopicPartition, long timestamp) {
     return RetryUtils.executeWithMaxAttemptAndExponentialBackoff(
         () -> pubSubConsumerAdapter.offsetForTime(pubSubTopicPartition, timestamp, kafkaOperationTimeout),
         25,
@@ -180,7 +183,10 @@ private Long offsetsForTimesUnprotected(PubSubTopicPartition pubSubTopicPartitio
         PUBSUB_RETRIABLE_FAILURES);
   }
 
-  private Long endOffsetsWithRetryUnprotected(PubSubTopicPartition partition) {
+  /**
+   * This method needs to be called with the pubSubConsumerLock held.
+   */
+  private Long fetchEndOffsetsWithRetry(PubSubTopicPartition partition) {
     return RetryUtils.executeWithMaxAttemptAndExponentialBackoff(
         () -> pubSubConsumerAdapter.endOffset(partition),
         25,
@@ -410,14 +416,14 @@ private void isTopicPartitionValid(PubSubTopicPartition pubSubTopicPartition) {
   public long getOffsetByTimeIfOutOfRange(PubSubTopicPartition pubSubTopicPartition, long timestamp) {
     isTopicPartitionValid(pubSubTopicPartition);
     try (AutoCloseableLock ignore = AutoCloseableLock.of(pubSubConsumerLock)) {
-      return getOffsetByTimeIfOutOfRangeUnprotected(pubSubTopicPartition, timestamp);
+      return fetchOffsetByTimeIfOutOfRange(pubSubTopicPartition, timestamp);
     }
   }
 
   /**
    * Call this function with the consumer lock held.
    */
-  private long getOffsetByTimeIfOutOfRangeUnprotected(PubSubTopicPartition pubSubTopicPartition, long timestamp)
+  private long fetchOffsetByTimeIfOutOfRange(PubSubTopicPartition pubSubTopicPartition, long timestamp)
       throws PubSubTopicDoesNotExistException {
     long latestOffset = getLatestOffset(pubSubTopicPartition);
     if (latestOffset <= 0) {