Kafka 6884 consumer group command should use new admin client #5032

asasvari · 2018-05-17T16:55:28Z

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

ConsumerGroupCommand is updated to use the new AdminClient.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Executed existing unit tests: DescribeConsumerGroupTest, DeleteConsumerGroupTest, DeleteConsumerGroupsTest, ResetConsumerGroupOffsetTest), ListConsumerGroupTest
Created a kafka-console-consumer & manually verified bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group 1

Committer Checklist (excluded from commit message)

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

asasvari · 2018-05-22T09:27:16Z

retest this please

asasvari · 2018-05-23T05:19:48Z

retest this please

asasvari · 2018-05-24T13:28:34Z

@viktorsomogyi can you take a look? In the meantime, I am trying to fix the failing tests.

vahidhashemian

Thanks for the PR. I did a quick look and left some comments.

vahidhashemian · 2018-05-24T21:17:38Z

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

        val (partitionsToResetWithCommittedOffset, partitionsToResetWithoutCommittedOffset) =
          partitionsToReset.partition(currentCommittedOffsets.keySet.contains(_))

        val preparedOffsetsForPartitionsWithCommittedOffset = partitionsToResetWithCommittedOffset.map { topicPartition =>
          (topicPartition, new OffsetAndMetadata(currentCommittedOffsets.get(topicPartition) match {
-            case Some(offset) => offset
+            case offset => offset.offset()


This makes the next check (case _) unreachable. Perhaps use case offset if offset != null or refactor the block to better fit the Java map get semantics.

vahidhashemian · 2018-05-24T21:26:39Z

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

 import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer, OffsetAndMetadata}
-import org.apache.kafka.common.errors.BrokerNotAvailableException
-import org.apache.kafka.common.{KafkaException, Node, TopicPartition}
+import org.apache.kafka.common.errors.{BrokerNotAvailableException, CoordinatorNotAvailableException, TimeoutException}


Unused import TimeoutException?

vahidhashemian · 2018-05-24T21:26:55Z

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

-import org.apache.kafka.common.errors.BrokerNotAvailableException
-import org.apache.kafka.common.{KafkaException, Node, TopicPartition}
+import org.apache.kafka.common.errors.{BrokerNotAvailableException, CoordinatorNotAvailableException, TimeoutException}
+import org.apache.kafka.common.{KafkaException, KafkaFuture, Node, TopicPartition}


Unused import KafkaFuture?

vahidhashemian · 2018-05-24T21:28:07Z

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

@@ -267,6 +270,7 @@ object ConsumerGroupCommand extends Logging {
      val offsetsOptPresent = opts.options.has(opts.offsetsOpt)
      val subActions = Seq(membersOptPresent, offsetsOptPresent, stateOptPresent).count(_ == true)

+


nit: extra blank line

vahidhashemian · 2018-05-24T21:32:43Z

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

-      adminClient.listAllConsumerGroupsFlattened().map(_.groupId)
+      val result = adminClient.listConsumerGroups()
+      val listings = result.valid().get()
+      listings.map(_.groupId()).toList


nit: Any reason for not putting all this in one line?
adminClient.listConsumerGroups().valid().get().map(_.groupId()).toList

vahidhashemian · 2018-05-24T21:38:41Z

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

@@ -640,10 +659,10 @@ object ConsumerGroupCommand extends Logging {
      if (consumer != null) consumer.close()
    }

-    private def createAdminClient(): AdminClient = {
+    private def createAdminClient(): org.apache.kafka.clients.admin.AdminClient = {


nit: You could use an import like below
import org.apache.kafka.clients.admin.{AdminClient => NewAdminClient}
and use only NewAdminClient here (i.e. private def createAdminClient(): NewAdminClient).

Also few lines below (NewAdminClient.create(props))

vahidhashemian · 2018-05-24T21:51:02Z

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

-        case currentState =>
-          printError(s"Assignments can only be reset if the group '$groupId' is inactive, but the current state is $currentState.")
+      val t = opts.options.valueOf(opts.timeoutMsOpt).intValue()
+      val opt = new DescribeConsumerGroupsOptions().timeoutMs(t)


nit: Maybe a one-liner can for these two lines?
val opt = new DescribeConsumerGroupsOptions().timeoutMs(opts.options.valueOf(opts.timeoutMsOpt).intValue())

In my opinion it is more readable and easier to debug this way.
Extracted to a new function (it is used multiple times).

vahidhashemian · 2018-05-24T21:55:32Z

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

@@ -789,13 +820,13 @@ object ConsumerGroupCommand extends Logging {
          case (topicPartition, newOffset) => (topicPartition, new OffsetAndMetadata(newOffset))
        }
      } else if (opts.options.has(opts.resetToCurrentOpt)) {
-        val currentCommittedOffsets = adminClient.listGroupOffsets(groupId)
+        val currentCommittedOffsets = adminClient.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata().get()


nit: a few blocks above the same thing has been done in two lines, i.e.

val consumerGroupOffsets = adminClient.listConsumerGroupOffsets(groupId) val currentCommittedOffsets = consumerGroupOffsets.partitionsToOffsetAndMetadata().get()

They should ideally match.

ijuma · 2018-05-26T15:51:49Z

Shall we take the chance and remove the methods from the internal Scala AdminClient?

asasvari · 2018-05-28T09:07:13Z

@ijuma I checked that in order to safely delete the methods, we need to adjust other tests, such as LegacyAdminClientTest, too. I would rather remove those methods from the internal Scala AdminClient in a follow-up JIRA. Doing so would keep this PR relatively small and easy to review.

andrasbeni · 2018-05-28T09:12:06Z

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

 import org.I0Itec.zkclient.exception.ZkNoNodeException
 import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.clients.admin.{AdminClient => NewAdminClient}


@asasvari Other commands import o.a.k.clients.admin.AdminClient as JAdminClient. I believe being consistent in naming helps others understand and maintain this piece of code.

viktorsomogyi · 2018-05-28T09:04:08Z

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

+        case e : ExecutionException if e.getCause.isInstanceOf[KafkaException] =>
+          printError(s"listGroup failed. Error message: " + e.getMessage)
+          if (e.getCause != null) {
+            throw new KafkaException(e.getCause.getCause.getMessage, e.getCause.getCause) // authentication exception


Are we keeping backward compatibility here by throwing the cause of cause? Is it always an AuthenticationException?

viktorsomogyi · 2018-05-28T10:00:02Z

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

 import scala.collection.JavaConverters._
 import scala.collection.{Seq, Set, mutable}
+import scala.concurrent.ExecutionException
+import scala.util.{Failure, Success, Try}

 object ConsumerGroupCommand extends Logging {


Would you please add tests that list, describe, etc. works with an empty string as group id? Currently it is possible to use an empty string as a group id in case of KafkaConsumer.assign() (but not with KafkaConsumer.subscribe().

According to GroupCoordinator empty group id is invalid.

Anyway, I tried to create a addConsumerGroupExecutor(numConsumers = 1, group = "") in DescribeConsumerGroupTest join fails with INVALID_GROUP_ID error:

[2018-05-29 15:35:48,754] ERROR [Consumer clientId=consumer-1, groupId=] Attempt to join group failed due to fatal error: The configured groupId is invalid (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:538)

Can you give me some hints @viktorsomogyi ?

viktorsomogyi · 2018-05-28T11:27:33Z

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

      val props = if (opts.options.has(opts.commandConfigOpt)) Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)) else new Properties()
      props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt))
-      AdminClient.create(props)
+      org.apache.kafka.clients.admin.AdminClient.create(props)


NewAdminClient.create(props)?

asasvari · 2018-05-28T14:37:54Z

@hachikuji I rebased the patch to the latest commit b7513214860008070568eae2f28b7eb581166d72. I can see there is a commit "MINOR: Remove MaxPermSize from gradle.properties" 7f19df29ac60ec9aef1dbd14704a2dd045b1cfbf that affects my patch, and timeout related tests such as testDescribeGroupOffsetsWithShortInitializationTimeout in DescribeConsumerGroupTest now fail with The consumer group command should fail due to low initialization timeout. It looks like retry backoff changes "disable" the --timeout argument. I tested that if I revert the commit 7f19df2, tests pass. Can you give me some hints why this could happen?

ijuma · 2018-05-29T14:25:17Z

@asasvari, it's fine to remove the methods in a separate PR.

ijuma · 2018-05-29T14:28:42Z

Also, to clarify, @asasvari is stating that the retry backoff changes (7f19df2) are causing some of the tests to fail, not the MaxPermSize changes.

asasvari · 2018-06-01T07:47:30Z

retest this please

asasvari · 2018-06-04T08:19:16Z

retest this please

asasvari · 2018-06-04T19:06:26Z

@ijuma, @hachikuji I addressed most of the review comments above. Can you or a committer take a look at the changes?

The JDK 8 tests have timed out; it does not look related to my changes (ConsumerGroupCommand tests passed).

16:17:32 kafka.api.PlaintextConsumerTest > testLowMaxFetchSizeForRequestAndPartition STARTED
17:43:26 Build timed out (after 180 minutes). Marking the build as aborted.
17:43:27 Build was aborted
17:43:27 [FINDBUGS] Skipping publisher since build result is ABORTED

asasvari · 2018-06-06T10:38:59Z

retest this please

vahidhashemian

I left some minor comments inline.

vahidhashemian · 2018-06-06T17:26:26Z

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

 import org.I0Itec.zkclient.exception.ZkNoNodeException
 import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.clients.admin.{AbstractOptions, DeleteConsumerGroupsOptions, DescribeConsumerGroupsOptions, ListConsumerGroupOffsetsOptions, ListConsumerGroupsOptions, AdminClient => JAdminClient}


Quoting @juma from another PR "The Scala AdminClient is going away and there is no need to have this rename in classes that don't need it."
You could do import org.apache.kafka.clients.admin and use admin.AdminClient to avoid the name clash.

vahidhashemian · 2018-06-06T17:32:21Z

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

@@ -42,8 +42,11 @@ import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.apache.kafka.common.utils.Utils

+import scala.collection.JavaConversions._


JavaConversions is deprecated starting as of Scala 2.12. You should use JavaConverters instead.
In a recent PR all usages of JavaConversions were removed.

vahidhashemian · 2018-06-06T17:46:29Z

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

-      )
+      val groupId = opts.options.valueOf(opts.groupOpt)
+      val consumerGroups = adminClient.describeConsumerGroups(
+        util.Arrays.asList(groupId),


nit: This can be done as List(groupId).asJava using JavaConverters (no need to import java.util). There are a couple of other instances of this.

vahidhashemian · 2018-06-06T17:49:18Z

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

      }

-      (Some(consumerGroupSummary.state), assignments)
+      val assignments = rowsWithConsumer ++ rowsWithoutConsumer
+      (Some(state.toString), Some(assignments))


nit: a one liner works too: (Some(state.toString), Some(rowsWithConsumer ++ rowsWithoutConsumer))

vahidhashemian · 2018-06-06T17:52:22Z

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

+      val consumerGroups = adminClient.describeConsumerGroups(
+        util.Arrays.asList(groupId),
+        withTimeoutMs(new DescribeConsumerGroupsOptions))
+        .describedGroups()


personal opinion: this is more readable written like this

val consumerGroups = adminClient.describeConsumerGroups( util.Arrays.asList(groupId), withTimeoutMs(new DescribeConsumerGroupsOptions) ).describedGroups()

Noticed this in a few places.

vahidhashemian · 2018-06-06T17:55:46Z

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

        val (partitionsToResetWithCommittedOffset, partitionsToResetWithoutCommittedOffset) =
          partitionsToReset.partition(currentCommittedOffsets.keySet.contains(_))

        val preparedOffsetsForPartitionsWithCommittedOffset = partitionsToResetWithCommittedOffset.map { topicPartition =>
          (topicPartition, new OffsetAndMetadata(currentCommittedOffsets.get(topicPartition) match {
-            case Some(offset) => offset
+            case offset if offset != null=> offset.offset()


nit: space before =>

vahidhashemian · 2018-06-06T18:25:38Z

core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala

@@ -19,10 +19,11 @@ package kafka.admin
 import kafka.utils.TestUtils
 import org.apache.kafka.clients.consumer.RoundRobinAssignor
 import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.errors.TimeoutException
+import org.apache.kafka.common.errors.{CoordinatorNotAvailableException, TimeoutException}


nit: unused CoordinatorNotAvailableException?

vahidhashemian · 2018-06-06T18:26:35Z

core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala

@@ -12,7 +12,7 @@
  */
 package kafka.api

-import java.io.FileOutputStream
+import java.io.{BufferedWriter, FileOutputStream, FileWriter, PrintWriter}


nit: unused added imports

asasvari · 2018-06-08T07:24:01Z

@vahidhashemian thanks for the review!

asasvari · 2018-06-08T10:03:18Z

retest this please

hachikuji

Thanks for the patch! Left a few small comments.

hachikuji · 2018-06-25T23:07:31Z

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

+
+      try {
+        val listings = result.all().get().asScala
+        listings.map(_.groupId()).toList


nit: no need for parenthesis for getters like groupId

hachikuji · 2018-06-25T23:09:11Z

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

+        listings.map(_.groupId()).toList
+      }
+      catch {
+        case e : ExecutionException if e.getCause.isInstanceOf[KafkaException] =>


It's a little unclear why we have the custom exception handling here. Previously we just raised all errors? Should we raise the cause in all cases?

The way the new KafkaAdminClient works is a little bit different than previously. I have seen that certain kind of exceptions (e.g. SaslAuthenticationException) are now nested, and I wanted to keep similar behaviour.

hachikuji · 2018-06-25T23:11:59Z

core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala

-                        "--describe",
-                        "--group", "test.group",
-                        "--command-config", propsFile.getAbsolutePath)
+      "--describe",


nit: was this intentional?

hachikuji · 2018-06-25T23:16:45Z

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

+            Map.empty
+        }
+      } catch {
+        case e : ExecutionException if e.getCause.isInstanceOf[CoordinatorNotAvailableException] =>


Again it's a little unclear why COORDINATOR_NOT_AVAILABLE is handled specially. Note that with #5278, the admin client should handle this error internally.

hachikuji · 2018-06-25T23:18:01Z

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

@@ -456,6 +519,13 @@ object ConsumerGroupCommand extends Logging {
      }
    }

+    private def getPartitionsToOffsetAndMetadata(groupId: String) = {


Maybe a simpler name is getCommittedOffsets?

hachikuji · 2018-06-25T23:18:48Z

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

+      val rowsWithConsumer = if (offsetsAndMetadataByTopic.isEmpty) {
+        List[PartitionAssignmentState]()
+      } else {
+        consumerGroup.members().asScala.filter(!_.assignment().topicPartitions().isEmpty).toList.sortWith(


nit: unneeded parenthesis for getters

hachikuji · 2018-06-25T23:20:42Z

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

+      val result = deletedGroups.mapValues { f =>
+        Try(f.get()) match {
+          case _: Success[_] => Errors.NONE
+          case Failure(ee: ExecutionException) if ee.getMessage.contains("TimeoutException") => Errors.COORDINATOR_NOT_AVAILABLE


This is a bit weird. Can you explain the reasoning?

I tried to follow the logic of the previous version of this method:

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

Lines 590 to 599 in 9ea81ba

val successfullyDeleted = result.filter {

case (_, error) => error == Errors.NONE

}.keySet

if (successfullyDeleted.size == result.size)

println(s"Deletion of requested consumer groups (${successfullyDeleted.mkString("'", "', '", "'")}) was successful.")

else {

printError("Deletion of some consumer groups failed:")

result.foreach {

case (group, error) if error != Errors.NONE => println(s"* Group '$group' could not be deleted due to: ${error.toString}")

KafkaAdminClient now returns futures as part of DeleteConsumerGroupsResult when you try to delete with deleteConsumerGroups(). Please let me know if you have an idea how to simplify it.

I think the part that is unclear to me is why we are checking for TimeoutException in the exception message. It seems like a brittle solution. Can we make it simple and just treat all exceptions as failures?

hachikuji · 2018-06-25T23:21:42Z

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

-            if (verbose) consumer.assignment else List())
-        })
-      )
+      private[admin] def collectGroupMembers(verbose: Boolean): (Option[String], Option[Seq[MemberAssignmentState]]) = {


Seems misaligned? Same for collectGroupState.

hachikuji · 2018-06-25T23:23:55Z

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



 import kafka.utils._
-import kafka.utils.Implicits._
+import org.I0Itec.zkclient.exception.ZkNoNodeException


Did we need this import? There are a few more unneeded imports to clean up as well.

vahidhashemian

Thank you for the updates. I left a few minor comments.

vahidhashemian · 2018-06-25T21:49:47Z

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

 import javax.xml.datatype.DatatypeFactory
 import joptsimple.{OptionParser, OptionSpec}
+import kafka.common.{OffsetMetadataAndError, TopicAndPartition}


unused imports? Also an extra blank line below

vahidhashemian · 2018-06-25T21:50:30Z

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



 import kafka.utils._
-import kafka.utils.Implicits._
+import org.I0Itec.zkclient.exception.ZkNoNodeException


unused import?

vahidhashemian · 2018-06-25T21:51:00Z

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

 import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer, OffsetAndMetadata}
+import org.apache.kafka.common.errors.{BrokerNotAvailableException, CoordinatorNotAvailableException, SaslAuthenticationException}


BrokerNotAvailableException import does not seem to be used.

vahidhashemian · 2018-06-25T21:51:56Z

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

 import scala.collection.JavaConverters._
+import scala.collection.{Seq, Set, mutable}


unused/duplicate imports? I think this line can be safely removed.

vahidhashemian · 2018-06-25T22:29:20Z

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

+          printError(s"listGroup failed. Error message: " + e.getMessage)
+          if (e.getCause.getCause.isInstanceOf[SaslAuthenticationException]) {
+            throw e.getCause.getCause
+          }


I'm seeing duplicate / inconsistent error reporting, for example when the broker cannot be found:

Error: listGroup failed. Error message: org.apache.kafka.common.KafkaException: Failed to find brokers to send ListGroups Error: Executing consumer group command failed due to Failed to find brokers to send ListGroups

or

Error: listGroup failed. Error message: org.apache.kafka.common.errors.ClusterAuthorizationException: Error listing groups on devstack1:9092 (id: 0 rack: null) Error: Executing consumer group command failed due to Error listing groups on devstack1:9092 (id: 0 rack: null)

I think this can be improved.

I removed extra logging call.

Regarding the consistency of messages: returned error depends on a broker's reply to the listGroup operation sent by the new KafkaAdminClient (Failed to find brokers / Error listing groups on). So, it does not look 100% deterministic / depends on multiple factors what kind error message is returned by a broker. Can you suggest how to handle the specific examples / different exceptions?

I agree that "Executing consumer group command failed due to Failed to find " sounds a bit weird. I will replace it with something like: Executing consumer group command failed with error: "<ERROR_MESSAGE>"

vahidhashemian · 2018-06-25T22:51:40Z

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

+            Map(topicPartition -> Some(offset.offset())),
+              Some(MISSING_COLUMN_VALUE),
+              Some(MISSING_COLUMN_VALUE),
+              Some(MISSING_COLUMN_VALUE))


Incorrect indentation on these three lines?

hachikuji

Thanks for the updates. Left a few more comments.

hachikuji · 2018-07-02T22:29:28Z

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

+        listings.map(_.groupId).toList
+      }
+      catch {
+        case e : ExecutionException if e.getCause.isInstanceOf[KafkaException] =>


I'm not sure how precisely we need to maintain the error reporting. I'm wondering if we can just raise e.getCause for all instances of ExecutionException.

In this "special" case, it is not enough to raise e.getCause.

When there is a SASL authentication failure, listings.map(_.groupId).toList throws:

concurrent.ExecutionException --> org.apache.kafka.common.KafkaException: Failed to find brokers to send ListGroups --> org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed due to invalid credentials with SASL mechanism SCRAM-SHA-256

KafkaAdminClient throws an ExecutionException when we try to get all results of the returned futures of listGroups (called from testConsumerGroupServiceWithAuthenticationFailure). The cause of the ExecutionException is a org.apache.kafka.common.KafkaException "Failed to find brokers to send ListGroups". Its cause is the SaslAuthenticationException if there is an authentication failure.

hachikuji · 2018-07-02T22:30:05Z

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

@@ -126,7 +128,21 @@ object ConsumerGroupCommand extends Logging {
    private var consumer: KafkaConsumer[String, String] = _

    def listGroups(): List[String] = {
-      adminClient.listAllConsumerGroupsFlattened().map(_.groupId)
+      val result = adminClient.listConsumerGroups(
+        withTimeoutMs(new ListConsumerGroupsOptions()))


nit: unneeded parenthesis for empty constructor

hachikuji · 2018-07-02T22:33:43Z

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

+        }
+      } catch {
+        case _: Throwable =>
+          printError(s"Coordinator not available for the group '$groupId'.")


This is definitely one possible case, but are we sure exceptions can't be thrown in other cases? Also, we probably should pass the exception to printError rather than swallowing it.

hachikuji · 2018-07-02T22:35:11Z

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

-            collectConsumerAssignment(group, Some(consumerGroupSummary.coordinator), Seq(topicPartition),
-              Map(topicPartition -> Some(offset)), Some(MISSING_COLUMN_VALUE),
-              Some(MISSING_COLUMN_VALUE), Some(MISSING_COLUMN_VALUE))
+    def collectGroupOffsets(): (Option[String], Option[Seq[PartitionAssignmentState]]) = {


nit: probably useful to have a docstring describing the return value.

hachikuji · 2018-07-02T22:36:30Z

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

+
+      val state = consumerGroup.state()
+      val committedOffsets = getCommittedOffsets(groupId).asScala.toMap
+      var assignedTopicPartitions = Array[TopicPartition]()


Maybe better to use a ListBuffer or something like that.

hachikuji · 2018-07-02T22:37:03Z

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

+        withTimeoutMs(new DescribeConsumerGroupsOptions())
+      ).describedGroups().get(groupId).get()
+
+      val state = consumerGroup.state()


nit: unneeded parenthesis. there are a few of these

hachikuji · 2018-07-02T22:41:05Z

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

+      val rowsWithConsumer = if (committedOffsets.isEmpty) {
+        List[PartitionAssignmentState]()
+      } else {
+        consumerGroup.members.asScala.filter(!_.assignment.topicPartitions.isEmpty).toList.sortWith(


nit: the alignment is a little weird. How about this?

consumerGroup.members.asScala.filter(!_.assignment.topicPartitions.isEmpty).toSeq .sortWith(_.assignment.topicPartitions.size > _.assignment.topicPartitions.size) .flatMap { consumerSummary => val topicPartitions = consumerSummary.assignment.topicPartitions.asScala ...

hachikuji · 2018-07-02T22:43:07Z

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

-          printError(s"Assignments can only be reset if the group '$groupId' is inactive, but the current state is $currentState.")
-          Map.empty
-      }
+    private def withTimeoutMs [T <: AbstractOptions[T] ] (options : T) =  {


nit: unneeded space after AbstractOptions[T]

hachikuji · 2018-07-02T22:44:16Z

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

-          val currentOffset = currentCommittedOffsets.getOrElse(topicPartition,
-            throw new IllegalArgumentException(s"Cannot shift offset for partition $topicPartition since there is no current committed offset"))
+          val currentOffset = currentCommittedOffsets.asScala.getOrElse(topicPartition,
+            throw new IllegalArgumentException(s"Cannot shift offset for partition $topicPartition since there is no current committed offset")).offset()


nit: it's a little clearer to do use map before the call to getOrElse.

hachikuji · 2018-07-02T22:47:43Z

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

+      val result = deletedGroups.mapValues { f =>
+        Try(f.get()) match {
+          case _: Success[_] => Errors.NONE
+          case Failure(ee: ExecutionException) if ee.getMessage.contains("TimeoutException") => Errors.COORDINATOR_NOT_AVAILABLE


I think the part that is unclear to me is why we are checking for TimeoutException in the exception message. It seems like a brittle solution. Can we make it simple and just treat all exceptions as failures?

* Update ConsumerGroupCommand to use the new AdminClient.

asasvari · 2018-07-09T08:16:24Z

retest this please

asasvari · 2018-07-10T16:26:55Z

retest this please

…estUtils

asasvari · 2018-07-11T13:55:15Z

Jenkins jobs failed without an obvious reason. See:

I could not reproduce the issue surfaced in previous runs. There the problem was an CoordinatorLoadInProgressException was thrown by the admin client when listgroup was called. I added retry mechanism to waitUntilTrue in TestUtils, but this test failed again. In the latest PR added more information via logging to see where the exception is originated from (was the RetriableException actually retried?).

@hachikuji Please let me know if you have additional hints how to handle / track down sporadical test failures (e.g. when a broker returns CoordinatorLoadInProgressException).

asasvari · 2018-07-11T13:55:28Z

retest this please

asasvari · 2018-07-12T10:26:48Z

retest this please

hachikuji

Sorry for the delay. Left a couple comments and a question about the error handling.

hachikuji · 2018-07-12T21:37:54Z

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

+        listings.map(_.groupId).toList
+      }
+      catch {
+        case e : ExecutionException if e.getCause.isInstanceOf[KafkaException] =>


nit: unneeded space before semicolon

hachikuji · 2018-07-12T21:39:06Z

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

+      catch {
+        case e : ExecutionException if e.getCause.isInstanceOf[KafkaException] =>
+          val kafkaException = e.getCause
+          if (kafkaException.getCause != null) {


Seems a little arbitrary to go one additional level. I'd suggest we just raise the cause of the ExecutionException.

By the way, this seems to be the only case where we worry about the ExecutionException. For other admin APIs, we just let it propagate. It would be nice to be consistent if possible. Going forward, I think we expect retriable errors to be handled internally in the AdminClient, which means only fatal errors and timeouts will reach this level. It may be useful to catch the timeouts and print a helpful message, but for anything else, we probably just want to raise the exception and fail the command. At least that seems like a conservative option until we've had a chance to refine the expected AdminClient error handling. So maybe we should remove this try/catch. What do you think?

Thanks for the clarification, I removed mentioned try-catch blocks. In general, I find it easier to read helpful error messages than stack traces, but you are right; KafkaAdminClient class has @InterfaceStability.Evolving annotation. Error handling can be refined in a separate PR.

hachikuji · 2018-07-12T21:42:23Z

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

+          throw kafkaException
+
+        case e: Exception =>
+          throw e


If we're just re-throwing the exception, then we can leave this case out.

hachikuji · 2018-07-12T21:45:22Z

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

+      } catch {
+        case ex: Throwable =>
+          val exceptionMessage = ex.getMessage
+          printError(s"Coordinator not available for the group '$groupId'. Error message: '$exceptionMessage'")


This seems a little dangerous if we're catching Throwable. Unexpected errors may not have a useful message and we would probably need the trace anyway. Similar to the comment above, maybe we can leave out this try/catch and let the exception propagate?

hachikuji

Thanks, looking pretty good. Just a couple more comments.

hachikuji · 2018-07-16T15:26:43Z

core/src/test/scala/unit/kafka/utils/TestUtils.scala

+    * @param msg error message
+    * @param waitTime maximum time to wait and retest the condition before failing the test
+    * @param pause delay between condition checks
+    * @param maxRetry maximum number of retries to check the given condition if a retriable exception is thrown


nit: maxRetries?

hachikuji · 2018-07-16T15:27:00Z

core/src/test/scala/unit/kafka/utils/TestUtils.scala

+      }
+      catch {
+        case _: RetriableException if retry < maxRetry => {
+          info("Retrying")


Maybe more useful to log the exception at DEBUG level? Also, maybe we should pause between retries?

hachikuji · 2018-07-16T15:39:17Z

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

+      val result = deletedGroups.mapValues { f =>
+        Try(f.get) match {
+          case _: Success[_] => Errors.NONE
+          case Failure(ee: ExecutionException) => Errors.forException(ee.getCause)


I'm not sure it's reasonable to assume we will have an ApiException. My concern is that we will lose information i the exception gets mapped to UNKNOWN_ERROR. I'm wondering if we could skip the translation to error code and use the exception directly.

deleteGroups() returns a Map[String, Errors] -

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

Line 587 in b687660

def deleteGroups(): Map[String, Errors] = {

I can change it to return Map[String, Throwable] and update all related tests (such as DeleteConsumerGroupsTest that expects certain errors to happen e.g. Errors.GROUP_ID_NOT_FOUND). If I do this, output for a delete group failure would change a bit. For example:

Error: Deletion of some consumer groups failed: * Group 'test.group' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty

I will update the PR soon.

hachikuji

Thanks, LGTM

asasvari · 2018-07-17T08:57:06Z

retest this please

asasvari · 2018-07-17T13:43:56Z

@hachikuji thanks for the review!

hachikuji · 2018-07-17T18:56:40Z

@asasvari Thanks, merged to trunk. Note I made a trivial tweak to waitUntil.

By the way, I created https://issues.apache.org/jira/browse/KAFKA-7173, which seems to be the last use of the old AdminClient. Not sure you have any interest, but thought I'd mention it.

asasvari mentioned this pull request May 18, 2018

KAFKA-6868: Fix buffer underflow and expose group state in the consumer groups API #4980

Merged

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from 90ee120 to a601f0f Compare May 21, 2018 21:09

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from a601f0f to 106f122 Compare May 22, 2018 13:45

hachikuji self-assigned this May 24, 2018

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from 106f122 to e308e9d Compare May 24, 2018 16:07

vahidhashemian requested changes May 24, 2018

View reviewed changes

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from e308e9d to b447bee Compare May 28, 2018 08:01

andrasbeni reviewed May 28, 2018

View reviewed changes

viktorsomogyi reviewed May 28, 2018

View reviewed changes

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from b447bee to cc552e0 Compare May 28, 2018 14:18

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from cc552e0 to 025e753 Compare May 31, 2018 12:55

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from 025e753 to e83f906 Compare June 1, 2018 15:31

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from e83f906 to d5ebc65 Compare June 4, 2018 14:42

vahidhashemian reviewed Jun 6, 2018

View reviewed changes

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from d5ebc65 to cd8d0fe Compare June 8, 2018 07:15

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from c9dfdb9 to 253051a Compare June 12, 2018 10:00

hachikuji reviewed Jun 25, 2018

View reviewed changes

vahidhashemian reviewed Jun 25, 2018

View reviewed changes

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from b955820 to 77fe7a6 Compare June 26, 2018 09:44

hachikuji reviewed Jul 2, 2018

View reviewed changes

asasvari added 4 commits July 3, 2018 11:07

KAFKA-6884: ConsumerGroupCommand should use new AdminClient

1e94610

* Update ConsumerGroupCommand to use the new AdminClient.

Fix review comments

f3481c2

Fix SaslClientsWithInvalidCredentialsTest related compilation warnings

3b5bb84

Fix review comments

6f42eda

asasvari force-pushed the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch from 7949612 to 6f42eda Compare July 4, 2018 10:25

Fix review comments

33d3820

Log retrying if a retryable exception is thrown in waitUntilTrue in T…

5ef15fd

…estUtils

hachikuji reviewed Jul 12, 2018

View reviewed changes

Fix review comment

48686a0

hachikuji reviewed Jul 16, 2018

View reviewed changes

Fix review comments

e66bd03

hachikuji approved these changes Jul 16, 2018

View reviewed changes

Add retriable error to debug message in TestUtils

fac4c84

hachikuji merged commit 8ec8ec5 into apache:trunk Jul 17, 2018

asasvari mentioned this pull request Jul 19, 2018

KAFKA-6960: Remove the methods from the internal Scala AdminClient th… #5395

Closed

3 tasks

bob-barrett mentioned this pull request Aug 12, 2019

MINOR: Use max retries for consumer group tests to avoid flakiness #7186

Merged

3 tasks

asasvari deleted the KAFKA-6884_ConsumerGroupCommand_should_use_new_AdminClient branch October 19, 2022 21:06

		@@ -267,6 +270,7 @@ object ConsumerGroupCommand extends Logging {
		val offsetsOptPresent = opts.options.has(opts.offsetsOpt)
		val subActions = Seq(membersOptPresent, offsetsOptPresent, stateOptPresent).count(_ == true)

	val successfullyDeleted = result.filter {
	case (_, error) => error == Errors.NONE
	}.keySet

	if (successfullyDeleted.size == result.size)
	println(s"Deletion of requested consumer groups (${successfullyDeleted.mkString("'", "', '", "'")}) was successful.")
	else {
	printError("Deletion of some consumer groups failed:")
	result.foreach {
	case (group, error) if error != Errors.NONE => println(s"* Group '$group' could not be deleted due to: ${error.toString}")

		import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer, OffsetAndMetadata}
		import org.apache.kafka.common.errors.{BrokerNotAvailableException, CoordinatorNotAvailableException, SaslAuthenticationException}

		import scala.collection.JavaConverters._
		import scala.collection.{Seq, Set, mutable}

Kafka 6884 consumer group command should use new admin client #5032

Kafka 6884 consumer group command should use new admin client #5032

Conversation

asasvari commented May 17, 2018 • edited Loading

Committer Checklist (excluded from commit message)

asasvari commented May 22, 2018

asasvari commented May 23, 2018

asasvari commented May 24, 2018

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

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

ijuma commented May 26, 2018

asasvari commented May 28, 2018

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

asasvari commented May 28, 2018

ijuma commented May 29, 2018

ijuma commented May 29, 2018

asasvari commented Jun 1, 2018

asasvari commented Jun 4, 2018

asasvari commented Jun 4, 2018

asasvari commented Jun 6, 2018

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

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

asasvari commented Jun 8, 2018

asasvari commented Jun 8, 2018

hachikuji 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

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

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

hachikuji 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

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

asasvari commented May 17, 2018 •

edited

Loading