Migrating from AdminUtils with AdminClient #848

surajkn · 2021-08-10T16:36:47Z

Replace kafka.admin.AdminUtils usage with kafka.client.AdminClient because AdminUtils is deprecated.
Also in a following effort 101tec ZkClient will be replaced by helix ZkClient and this is required for that.

Important: DO NOT REPORT SECURITY ISSUES DIRECTLY ON GITHUB.
For reporting security issues and contributing security fixes,
please, email security@linkedin.com instead, as described in
the contribution guidelines.

Please, take a minute to review the contribution guidelines at:
https://github.com/linkedin/Brooklin/blob/master/CONTRIBUTING.md

surajkn · 2021-08-10T16:38:31Z

gradle/dependency-versions.gradle

@@ -21,4 +21,5 @@ ext {
    testngVersion = "7.1.0"
    zkclientVersion = "0.11"


I should probably delete this as well

If there are no usages you should

This is still required. Will remove this in a the following PR where I fully remove and replace 101tec ZkClient with Helix ZkClient.

jzakaryan

@surajkn did a walkthrough of the changes and explained the differences in APIs. Waiting for him to confirm config changes for Kafka client before approving the PR.

surajkn · 2021-08-20T18:07:57Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTestUtils.java

   * @param topic the topic to wait for broker assignment
   * @param brokerList the brokers in the Kafka cluster
   * @throws IllegalStateException if the topic is not ready before the timeout ({@value #DEFAULT_TIMEOUT_MS} ms)
   */
-  public static void waitForTopicCreation(ZkUtils zkUtils, String topic, String brokerList) throws IllegalStateException {
-    Validate.notNull(zkUtils);
+  public static void waitForTopicCreation(AdminClient adminClient, String topic, String brokerList) throws IllegalStateException {


I am not sure if we need this anymore. In this patch I replace calls to AdminUtils.createTopic with calls to AdminClient.createTopics method, and createTopics essentially returns a "KafkaFuture" object on which I currently do a blocking wait (to wait for topic to be created). So we probably don't need this anymore? I have currently removed all calls to this method in test cases and its working fine.

surajkn · 2021-08-20T18:11:16Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTestUtils.java

+   * @param adminClient AdminClient instance to check if topics exists
+   * @param topic Topic name
+   */
+  public static boolean topicExists(AdminClient adminClient, String topic) {


I tried to import "topicExists" method from BaseKafkaZkTest class here but I kept getting "package com.linkedin.datastream.testutil does not exist" error so I copied the method here. Its probably some classpath issue but I am not sure how to resolve it in this repo.

Is this because of the circular dependency problem we discussed offline?

Yes, correct.

If it's a circular dep issue, then could we reference this function from BaseKafkaZkTest class's call to avoid duplicate code?

I can check but I doubt that will work. I guess trying to do it either way will still result in circular dependency.

surajkn · 2021-08-20T18:14:33Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProviderAdmin.java

+      // Removing from topic config since its passed as a direct argument
+      topicProperties.remove("replicationFactor");
+      // TopicExistsException is thrown if topic exists. Its a no-op.
+      adminClient.createTopics(Collections.singletonList(newTopic)).all().get();


By doing ".all().get()" we have a blocking wait for the topic to be created. I am not sure what exactly was the behavior with AdminUtils and/or if doing a blocking wait is the correct thing to do here. I am guessing this is the correct way?

However, having said that. On repeated execution of all the unit tests, I noticed that the test "testCreateDatastreamHappyPathDefaultRetention" is a little flaky in that it randomly fails every now and then (very rarely though, hard to reproduce). This test calls "getRetention" which queries topic config and returns retention time if its exists in policy else null, and when this test fails its because getRetention returned null instead of actual expected value. I am not sure if this test was flaky even before. I am not sure how to explain this behavior.

Summarizing the offline discussion here:

There's no callbacks and returned futures with AdminUtils.createTopic. No hints that it's an asynchronous operation. I'm not a scala expert, but from the looks of it it's a synchronous call.

As for the failing/flaky test:

Was this test flaky even before or it became flaky as a result of the API changes? We need to rule out changes in behavior of the system after switching from AdminUtils to AdminClient

Does adding a thread sleep after topic creation (and before querying the retention of the new topic) make the test pass?

For (1), we'll reach out to Kafka to confirm that there are no changes in behavior in createTopic APIs. For (2) we can try fixing the flaky test in a separate PR.

surajkn · 2021-08-20T18:16:26Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProviderAdmin.java

      }
    } catch (Throwable e) {
      LOG.error("Creating topic {} failed with exception {}", topicName, e);
      throw e;
    }
  }

+  private AdminClient getAdminClient(Datastream datastream) {


Do we need to cache AdminClient instances based on the destinationBrokers ? Depends on how frequently this method is called and how expensive it is to re-instantiate AdminClient object, but I am not sure of it.

In an offline discussion with Jhora we decided to not do it for now and do it later if the need arises.

surajkn · 2021-08-20T18:17:26Z

datastream-kafka/src/test/java/com/linkedin/datastream/kafka/TestKafkaTransportProvider.java


-    KafkaTestUtils.waitForTopicCreation(_zkUtils, topicName, _kafkaCluster.getBrokers());
+    //KafkaTestUtils.waitForTopicCreation(_adminClient, topicName, _kafkaCluster.getBrokers());


Will remove this in following update to this patch. Same for another instance below.

jzakaryan · 2021-08-25T00:03:37Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProviderAdmin.java

+      // TopicExistsException is thrown if topic exists. Its a no-op.
+      adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
+    } catch (InterruptedException | ExecutionException e) {
+      if (e instanceof TopicExistsException || e.getCause() instanceof TopicExistsException) {


I may be missing something here, but org.apache.kafka.common.errors.TopicExistsException is neither an InterruptedException nor an ExecutionException according to https://kafka.apache.org/28/javadoc//org/apache/kafka/common/errors/TopicExistsException.html.
How can the first condition of this if statement ever be true? (and I know this code was there even before)

Actually I found this reference implementation in LiKafkaTransporterProviderAdmin implementation here
But let me double check this once anyway.

So looks like TopicExistsException has an inheritance hierarchy (chain) that inherits from java.lang.Exception and both InterruptedException and ExecutionException also inherit from same java.lang.Excepption.

I understand that, my point was that TopicExistsException is neither a descendant of InterruptedException nor is it a descendant of ExecutionException. That first condition is always false.

yea so, we may need recursive calls to get the cause of the exceptions until we hit the topic exists right?

So I suspect TopicExistsException is raised as "ExecutionException(new TopicExistsException())" and that is why it needs to be handled this way. If I split it and add a separate catch block for TopicExistsException then it fails to catch that exception and that exception goes uncaught. I tried it and saw a few unit tests fail because of it.

jzakaryan · 2021-08-25T00:27:55Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTestUtils.java

+   * @param adminClient AdminClient instance to check if topics exists
+   * @param topic Topic name
+   */
+  public static boolean topicExists(AdminClient adminClient, String topic) {


Is this because of the circular dependency problem we discussed offline?

jzakaryan · 2021-08-25T16:28:45Z

gradle/dependency-versions.gradle

@@ -21,4 +21,5 @@ ext {
    testngVersion = "7.1.0"
    zkclientVersion = "0.11"


If there are no usages you should

vmaheshw

Can you please split this PR and separate out the admin client change? AdminClient change can be merged as is.

For the zk client migration part, we can see if we have a jfrog branching. Otherwise, we can duplicate some of the code and have different classes called using config knob and then later get rid of the knob, if things are working fine. It may duplicate some code, but will definitely unblock you and remove your dependency from jfrog branching.

Undid the ZkClient changes to revert back to extending 101tec ZkClient

shrinandthakkar · 2021-08-31T17:05:43Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTestUtils.java

+   * @param adminClient AdminClient instance to check if topics exists
+   * @param topic Topic name
+   */
+  public static boolean topicExists(AdminClient adminClient, String topic) {


If it's a circular dep issue, then could we reference this function from BaseKafkaZkTest class's call to avoid duplicate code?

...java/com/linkedin/datastream/connectors/kafka/mirrormaker/TestKafkaMirrorMakerConnector.java

datastream-testcommon/src/main/java/com/linkedin/datastream/testutil/BaseKafkaZkTest.java

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProviderAdmin.java

shrinandthakkar · 2021-08-31T17:26:52Z

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProviderAdmin.java

+      // TopicExistsException is thrown if topic exists. Its a no-op.
+      adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
+    } catch (InterruptedException | ExecutionException e) {
+      if (e instanceof TopicExistsException || e.getCause() instanceof TopicExistsException) {


yea so, we may need recursive calls to get the cause of the exceptions until we hit the topic exists right?

Replace kafka.admin.AdminUtils usage with kafka.client.AdminClient because AdminUtils is deprecated. Also in a following effort 101tec ZkClient will be replaced by helix ZkClient and this is required for that.

jzakaryan

LGTM

Replace kafka.admin.AdminUtils usage with kafka.client.AdminClient because AdminUtils is deprecated. Also in a following effort 101tec ZkClient will be replaced by helix ZkClient and this is required for that.

surajkn requested review from vmaheshw, jzakaryan and shrinandthakkar August 10, 2021 16:36

surajkn commented Aug 10, 2021

View reviewed changes

jzakaryan reviewed Aug 12, 2021

View reviewed changes

surajkn force-pushed the 101tec_zkclient_to_helix_zkclient branch from 90983f2 to 4348b9d Compare August 20, 2021 17:57

surajkn commented Aug 20, 2021

View reviewed changes

jzakaryan reviewed Aug 25, 2021

View reviewed changes

vmaheshw previously requested changes Aug 26, 2021

View reviewed changes

surajkn force-pushed the 101tec_zkclient_to_helix_zkclient branch from 4348b9d to 7eb32d2 Compare August 31, 2021 05:34

surajkn changed the title ~~Migrating from 101tec ZkClient to Helix ZkClient~~ Migrating from AdminUtils with AdminClient Aug 31, 2021

surajkn requested a review from somandal August 31, 2021 16:43

shrinandthakkar reviewed Aug 31, 2021

View reviewed changes

surajkn force-pushed the 101tec_zkclient_to_helix_zkclient branch from 7eb32d2 to 9a12243 Compare September 8, 2021 23:32

jzakaryan self-requested a review September 9, 2021 23:16

jzakaryan previously approved these changes Sep 9, 2021

View reviewed changes

Migrating from AdminUtils with AdminClient

3aca65b

Replace kafka.admin.AdminUtils usage with kafka.client.AdminClient because AdminUtils is deprecated. Also in a following effort 101tec ZkClient will be replaced by helix ZkClient and this is required for that.

surajkn dismissed jzakaryan’s stale review via 3aca65b September 17, 2021 02:28

surajkn force-pushed the 101tec_zkclient_to_helix_zkclient branch from 9a12243 to 3aca65b Compare September 17, 2021 02:28

shrinandthakkar approved these changes Sep 17, 2021

View reviewed changes

jzakaryan approved these changes Sep 20, 2021

View reviewed changes

surajkn merged commit 5ccd661 into linkedin:master Sep 21, 2021

surajkn deleted the 101tec_zkclient_to_helix_zkclient branch September 21, 2021 16:06

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Migrating from AdminUtils with AdminClient #848

Migrating from AdminUtils with AdminClient #848

surajkn commented Aug 10, 2021 •

edited

Loading

surajkn Aug 10, 2021

jzakaryan Aug 25, 2021

surajkn Sep 8, 2021

jzakaryan left a comment

surajkn Aug 20, 2021

surajkn Aug 20, 2021

jzakaryan Aug 25, 2021

surajkn Aug 25, 2021

shrinandthakkar Aug 31, 2021

surajkn Sep 1, 2021

surajkn Aug 20, 2021

jzakaryan Aug 31, 2021

surajkn Aug 20, 2021

surajkn Sep 8, 2021

surajkn Aug 20, 2021

surajkn Sep 8, 2021

jzakaryan Aug 25, 2021

surajkn Aug 25, 2021

surajkn Aug 25, 2021

jzakaryan Aug 31, 2021

shrinandthakkar Aug 31, 2021

surajkn Sep 8, 2021

jzakaryan Aug 25, 2021

jzakaryan Aug 25, 2021

vmaheshw left a comment

shrinandthakkar Aug 31, 2021

shrinandthakkar Aug 31, 2021

jzakaryan left a comment

		@@ -21,4 +21,5 @@ ext {
		testngVersion = "7.1.0"
		zkclientVersion = "0.11"


		KafkaTestUtils.waitForTopicCreation(_zkUtils, topicName, _kafkaCluster.getBrokers());
		//KafkaTestUtils.waitForTopicCreation(_adminClient, topicName, _kafkaCluster.getBrokers());

Migrating from AdminUtils with AdminClient #848

Migrating from AdminUtils with AdminClient #848

Conversation

surajkn commented Aug 10, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jzakaryan 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

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

vmaheshw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jzakaryan left a comment

Choose a reason for hiding this comment

surajkn commented Aug 10, 2021 •

edited

Loading