KAFKA-6321: Consolidate calls to KafkaConsumer's `beginningOffsets()` and `endOffsets()` in ConsumerGroupCommand #4344

vahidhashemian · 2017-12-19T21:05:46Z

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

tedyu · 2017-12-19T21:45:26Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

+    protected def getLogEndOffsets(topicPartitions: Seq[TopicPartition]): Map[TopicPartition, LogOffsetResult] = {
+      val offsets = getConsumer.endOffsets(topicPartitions.asJava)
+      topicPartitions.map { topicPartition =>
+        val logStartOffset = offsets.get(topicPartition)


Should this variable be renamed to match the method name ?

The method is getLogEndOffsets.

Yes, thanks for catching it. I'll submit an update.

vahidhashemian · 2017-12-20T04:24:46Z

@becketqin This PR addresses the JIRA you opened. I'd appreciate your review when you get a chance. Thanks.

becketqin · 2017-12-20T21:03:53Z

Thanks for the patch. I am a little busy today. Will review tomorrow.

becketqin

Thanks for the patch. Please see some comments below.

becketqin · 2017-12-22T14:40:42Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

@@ -274,7 +274,10 @@ object ConsumerGroupCommand extends Logging {

    protected def opts: ConsumerGroupCommandOptions

-    protected def getLogEndOffset(topicPartition: TopicPartition): LogOffsetResult
+    protected def getLogEndOffset(topicPartition: TopicPartition): LogOffsetResult =
+      getLogEndOffsets(Seq(topicPartition)).get(topicPartition).getOrElse(LogOffsetResult.Ignore)


Could just be getLogEndOffsets(Seq(topicPartition)).getOrElse(topicPartition, LogOffsetResult.Ignore)

becketqin · 2017-12-22T14:48:50Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

@@ -274,7 +274,10 @@ object ConsumerGroupCommand extends Logging {

    protected def opts: ConsumerGroupCommandOptions

-    protected def getLogEndOffset(topicPartition: TopicPartition): LogOffsetResult
+    protected def getLogEndOffset(topicPartition: TopicPartition): LogOffsetResult =


It seems this method is still used in a loop in line 706 and line 771. Should we also replace them with batch call?

You're right. I'll try to consolidate this in the next commit. Thanks for catching it.

vahidhashemian

@becketqin, thanks for reviewing the PR. I tried to address your comments in the new commit.

vahidhashemian · 2017-12-22T20:28:07Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

@@ -274,7 +274,10 @@ object ConsumerGroupCommand extends Logging {

    protected def opts: ConsumerGroupCommandOptions

-    protected def getLogEndOffset(topicPartition: TopicPartition): LogOffsetResult
+    protected def getLogEndOffset(topicPartition: TopicPartition): LogOffsetResult =


You're right. I'll try to consolidate this in the next commit. Thanks for catching it.

becketqin

@vahidhashemian Thanks for updating the patch. Happy new year! I left a few more comments.

becketqin · 2018-01-02T12:20:46Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

-      val offsets = getConsumer.beginningOffsets(List(topicPartition).asJava)
-      val logStartOffset = offsets.get(topicPartition)
-      LogOffsetResult.LogOffset(logStartOffset)
+    protected def getLogStartOffset(topicPartition: TopicPartition): LogOffsetResult =


It seems this method is no longer used. Can we just delete it?

That's true. I'll remove it in the next commit. Thanks!

becketqin · 2018-01-02T12:25:29Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

+      topicPartitions.map { topicPartition =>
+        val logStartOffset = offsets.get(topicPartition)
+        topicPartition -> LogOffsetResult.LogOffset(logStartOffset)
+      }.toMap
    }

    protected def getLogTimestampOffset(topicPartition: TopicPartition, timestamp: java.lang.Long): LogOffsetResult = {


This method is called when resetting offsets by timestamp and duration. It seems we are still doing the reset partition by partition. Can we also batch them?

Yes, I'll do that in the next commit.

becketqin · 2018-01-02T12:35:34Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

-        case LogOffsetResult.LogOffset(endOffset) if offset > endOffset =>
-          warn(s"New offset ($offset) is higher than latest offset. Value will be set to $endOffset")
-          endOffset
+    private def checkOffsetRange(requestedOffsets: Map[TopicPartition, Long]) = {


Nit: perhaps we should change the method name to checkOffsetsRange(), with plural form.

Sure, will update this too.

vahidhashemian

@becketqin Happy new year! Thanks for another round of feedback. I tried to address them in the new commit.

vahidhashemian · 2018-01-02T20:16:55Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

-      val offsets = getConsumer.beginningOffsets(List(topicPartition).asJava)
-      val logStartOffset = offsets.get(topicPartition)
-      LogOffsetResult.LogOffset(logStartOffset)
+    protected def getLogStartOffset(topicPartition: TopicPartition): LogOffsetResult =


That's true. I'll remove it in the next commit. Thanks!

vahidhashemian · 2018-01-02T20:17:14Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

+      topicPartitions.map { topicPartition =>
+        val logStartOffset = offsets.get(topicPartition)
+        topicPartition -> LogOffsetResult.LogOffset(logStartOffset)
+      }.toMap
    }

    protected def getLogTimestampOffset(topicPartition: TopicPartition, timestamp: java.lang.Long): LogOffsetResult = {


Yes, I'll do that in the next commit.

vahidhashemian · 2018-01-02T20:17:56Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

-        case LogOffsetResult.LogOffset(endOffset) if offset > endOffset =>
-          warn(s"New offset ($offset) is higher than latest offset. Value will be set to $endOffset")
-          endOffset
+    private def checkOffsetRange(requestedOffsets: Map[TopicPartition, Long]) = {


Sure, will update this too.

vahidhashemian · 2018-01-18T21:19:16Z

@becketqin any other comment/feedback on this PR? Thanks.

becketqin · 2018-01-19T20:03:37Z

@vahidhashemian Sorry I missed the PR update. Will look at it later today.

becketqin

@vahidhashemian Thanks for updating the patch. Made another pass and left some comments.

becketqin · 2018-01-20T16:47:45Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

+      consumer.assign(topicPartitions.asJava)
+      val offsetsForTimes = consumer.offsetsForTimes(topicPartitions.map(_ -> timestamp).toMap.asJava)
+
+      if (offsetsForTimes == null || offsetsForTimes.isEmpty)


KafkaConsumer.offsetsForTimes() will always return a non-empty map as long as the initial partition list is not empty. If the offset of a particular partition is not available, the returned map will have that partition mapped to null.

Perhaps we can do the following here:

use offsetsForTimes() to get the result for all the specified partitions.

If there are partitions with null value in the map returned in step 1, add all those unsuccessful partitions to another set.

use getEndOffsets() to get the log end offsets for the unsuccessful partitions in step 2 and map those partitions to their log end offsets.

return the combined result from step 1 and 3.

Thanks, I made this change in the latest commit. It should save some cpu cycles.

becketqin · 2018-01-20T16:57:02Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

        }.toMap
+        checkOffsetsRange(requestedOffsets).map {
+          case (topicPartition, newOffset) => (topicPartition, new OffsetAndMetadata(newOffset))
+        }
      } else if (opts.options.has(opts.resetToCurrentOpt)) {
        val currentCommittedOffsets = adminClient.listGroupOffsets(groupId)
        partitionsToReset.map { topicPartition =>


Can we also batch the requests here? Perhaps we can add all the partitions without a committed offsets to a separate set and invoke the getLogEndOffsets() for those partitions.

Sure, that makes sense too. I've done that in the latest commit.

becketqin · 2018-01-20T17:00:56Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

+            case _ => offset
+          }
+
+          case None => // the control should not reach here


If the control should not reach here, maybe we can throw an IllegalStateException?

vahidhashemian

@becketqin Thanks for another look at this PR, and the suggestions. I tried to address them (additional comments inline) in the recent commit.

vahidhashemian · 2018-01-22T21:03:23Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

+      consumer.assign(topicPartitions.asJava)
+      val offsetsForTimes = consumer.offsetsForTimes(topicPartitions.map(_ -> timestamp).toMap.asJava)
+
+      if (offsetsForTimes == null || offsetsForTimes.isEmpty)


Thanks, I made this change in the latest commit. It should save some cpu cycles.

vahidhashemian · 2018-01-22T21:48:11Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

        }.toMap
+        checkOffsetsRange(requestedOffsets).map {
+          case (topicPartition, newOffset) => (topicPartition, new OffsetAndMetadata(newOffset))
+        }
      } else if (opts.options.has(opts.resetToCurrentOpt)) {
        val currentCommittedOffsets = adminClient.listGroupOffsets(groupId)
        partitionsToReset.map { topicPartition =>


Sure, that makes sense too. I've done that in the latest commit.

becketqin

Thanks for the latest patch. LGTM. Just one minor comment.

becketqin · 2018-01-23T20:42:02Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

+        val preparedOffsetsForParititionsWithCommittedOffset = partitionsToResetWithCommittedOffset.map { topicPartition =>
+          (topicPartition, new OffsetAndMetadata(currentCommittedOffsets.get(topicPartition) match {
+            case Some(offset) => offset
+            case _ => throw new IllegalStateException(s"Expected a valid current offset for topic partition: $topicPartition")


Currently we print the help message and exit. Is this change intentional?

This is another part of the code that should be reachable (because at this point we're certain currentCommittedOffsets.get(topicPartition) is not None). I used this instead of doing a currentCommittedOffsets.get(topicPartition).get above. If you prefer that one instead, or can think of a better way to go about it (that avoids using .get on an Option) please let me know and I'll make the change. Thanks.

Ah, this is for the partitions with committed offsets. This is fine.

becketqin · 2018-01-25T01:37:46Z

@vahidhashemian It looks there is a minor conflict when I was trying to check in the patch. Do you have time to rebase the patch? Thanks.

… and `endOffsets()` in ConsumerGroupCommand

This is to avoid repeated calls to `getLogStartOffset` and `getLogEndOffset`.

vahidhashemian · 2018-01-25T03:13:21Z

@becketqin Thanks for letting me know. The rebase is complete.

becketqin · 2018-01-26T04:42:17Z

@vahidhashemian Thanks for the patch. I just merged it to the trunk.
BTW, I found another related issue when testing the patch and created KAFKA-6489.

vahidhashemian · 2018-01-26T04:43:26Z

@becketqin, thanks! Let me know if you want me to work on that too.

becketqin · 2018-01-26T04:54:17Z

@vahidhashemian You are welcome. I have submitted a patch for KAFKA-6489. It is a small fix.

tedyu reviewed Dec 19, 2017

View reviewed changes

becketqin reviewed Dec 22, 2017

View reviewed changes

vahidhashemian commented Dec 22, 2017

View reviewed changes

vahidhashemian force-pushed the KAFKA-6321 branch 2 times, most recently from c6ca83c to 5fd728d Compare December 23, 2017 06:15

becketqin reviewed Jan 2, 2018

View reviewed changes

vahidhashemian commented Jan 2, 2018

View reviewed changes

becketqin reviewed Jan 20, 2018

View reviewed changes

vahidhashemian force-pushed the KAFKA-6321 branch from e24cdd6 to a55e0c6 Compare January 22, 2018 22:23

vahidhashemian commented Jan 22, 2018

View reviewed changes

becketqin reviewed Jan 23, 2018

View reviewed changes

vahidhashemian added 5 commits January 24, 2018 19:01

KAFKA-6321: Consolidate calls to KafkaConsumer's beginningOffsets()…

8582222

… and `endOffsets()` in ConsumerGroupCommand

Fix a variable name

0cae3f9

Perform consolidation for checkOffsetRange too

8add214

This is to avoid repeated calls to `getLogStartOffset` and `getLogEndOffset`.

Perform consolidation for getLogTimestampOffset too

6e8c6a4

Did a couple of more consolidations

3f60d6d

vahidhashemian force-pushed the KAFKA-6321 branch from a55e0c6 to 3f60d6d Compare January 25, 2018 03:11

becketqin closed this in 138f57f Jan 26, 2018

KAFKA-6321: Consolidate calls to KafkaConsumer's beginningOffsets() and endOffsets() in ConsumerGroupCommand #4344

KAFKA-6321: Consolidate calls to KafkaConsumer's beginningOffsets() and endOffsets() in ConsumerGroupCommand #4344

Conversation

vahidhashemian commented Dec 19, 2017 • edited by becketqin Loading

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian commented Dec 20, 2017

becketqin commented Dec 20, 2017

becketqin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

becketqin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian commented Jan 18, 2018

becketqin commented Jan 19, 2018

becketqin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

becketqin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

becketqin commented Jan 25, 2018

vahidhashemian commented Jan 25, 2018

becketqin commented Jan 26, 2018

vahidhashemian commented Jan 26, 2018

becketqin commented Jan 26, 2018

KAFKA-6321: Consolidate calls to KafkaConsumer's `beginningOffsets()` and `endOffsets()` in ConsumerGroupCommand #4344

KAFKA-6321: Consolidate calls to KafkaConsumer's `beginningOffsets()` and `endOffsets()` in ConsumerGroupCommand #4344

vahidhashemian commented Dec 19, 2017 •

edited by becketqin

Loading