kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [kafka] branch trunk updated: KAFKA-6916; Refresh metadata in admin client if broker connection fails (#5050)
Date Tue, 29 May 2018 15:37:32 GMT
This is an automated email from the ASF dual-hosted git repository.

jgus pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 3a8d3a7  KAFKA-6916; Refresh metadata in admin client if broker connection fails
(#5050)
3a8d3a7 is described below

commit 3a8d3a792755cb2bc2a72d3a64dd84b6c8653035
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
AuthorDate: Tue May 29 16:37:17 2018 +0100

    KAFKA-6916; Refresh metadata in admin client if broker connection fails (#5050)
    
    Refresh metadata if broker connection fails so that new calls are sent only to nodes that
are alive and requests to controller are sent to the new controller if controller changes
due to broker failure. Also reassign calls that could not be sent.
    
    Reviewers: Dong Lin <lindong28@gmail.com>, Jason Gustafson <jason@confluent.io>
---
 .../kafka/clients/admin/KafkaAdminClient.java      | 28 +++++++++++++++++++++-
 .../admin/internals/AdminMetadataManager.java      |  2 +-
 .../kafka/api/AdminClientIntegrationTest.scala     | 15 ++++++++++++
 3 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 6c35f00..793e1da 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -991,6 +991,29 @@ public class KafkaAdminClient extends AdminClient {
             }
         }
 
+        /**
+         * Reassign calls that have not yet been sent. When metadata is refreshed,
+         * all unsent calls are reassigned to handle controller change and node changes.
+         * When a node is disconnected, all calls assigned to the node are reassigned.
+         *
+         * @param now The current time in milliseconds
+         * @param disconnectedOnly Reassign only calls to nodes that were disconnected
+         *                         in the last poll
+         */
+        private void reassignUnsentCalls(long now, boolean disconnectedOnly) {
+            ArrayList<Call> pendingCallsToSend = new ArrayList<>();
+            for (Iterator<Map.Entry<Node, List<Call>>> iter = callsToSend.entrySet().iterator();
iter.hasNext(); ) {
+                Map.Entry<Node, List<Call>> entry = iter.next();
+                if (!disconnectedOnly || client.connectionFailed(entry.getKey())) {
+                    for (Call call : entry.getValue()) {
+                        pendingCallsToSend.add(call);
+                    }
+                    iter.remove();
+                }
+            }
+            chooseNodesForPendingCalls(now, pendingCallsToSend.iterator());
+        }
+
         private boolean hasActiveExternalCalls(Collection<Call> calls) {
             for (Call call : calls) {
                 if (!call.isInternal()) {
@@ -1075,6 +1098,7 @@ public class KafkaAdminClient extends AdminClient {
 
                 // Update the current time and handle the latest responses.
                 now = time.milliseconds();
+                reassignUnsentCalls(now, true); // reassign calls to disconnected nodes
                 handleResponses(now, responses);
             }
             int numTimedOut = 0;
@@ -1158,7 +1182,9 @@ public class KafkaAdminClient extends AdminClient {
                 @Override
                 public void handleResponse(AbstractResponse abstractResponse) {
                     MetadataResponse response = (MetadataResponse) abstractResponse;
-                    metadataManager.update(response.cluster(), time.milliseconds());
+                    long now = time.milliseconds();
+                    metadataManager.update(response.cluster(), now);
+                    reassignUnsentCalls(now, false);
                 }
 
                 @Override
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java
b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java
index e06aed2..85d3c28 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java
@@ -116,7 +116,7 @@ public class AdminMetadataManager {
 
         @Override
         public void requestUpdate() {
-            // Do nothing
+            AdminMetadataManager.this.requestUpdate();
         }
     }
 
diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
index 24daa86..231b1e7 100644
--- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
@@ -219,6 +219,21 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with
Logging {
     waitForTopics(client, List(), topics)
   }
 
+  @Test
+  def testMetadataRefresh(): Unit = {
+    client = AdminClient.create(createConfig())
+    val topics = Seq("mytopic")
+    val newTopics = Seq(new NewTopic("mytopic", 3, 3))
+    client.createTopics(newTopics.asJava).all.get()
+    waitForTopics(client, expectedPresent = topics, expectedMissing = List())
+
+    val controller = servers.find(_.config.brokerId == TestUtils.waitUntilControllerElected(zkClient)).get
+    controller.shutdown()
+    controller.awaitShutdown()
+    val topicDesc = client.describeTopics(topics.asJava).all.get()
+    assertEquals(topics.toSet, topicDesc.keySet.asScala)
+  }
+
   /**
     * describe should not auto create topics
     */

-- 
To stop receiving notification emails like this one, please contact
jgus@apache.org.

Mime
View raw message