KAFKA-8933; Fix NPE in DefaultMetadataUpdater after authentication failure

clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java

@@ -18,13 +18,13 @@
 
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.requests.MetadataResponse;
 import org.apache.kafka.common.requests.RequestHeader;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 /**
  * A simple implementation of `MetadataUpdater` that returns the cluster nodes set via the constructor or via
@@ -36,9 +36,6 @@
  * This class is not thread-safe!
  */
 public class ManualMetadataUpdater implements MetadataUpdater {
-
-    private static final Logger log = LoggerFactory.getLogger(ManualMetadataUpdater.class);
-
     private List<Node> nodes;
 
     public ManualMetadataUpdater() {
@@ -69,24 +66,18 @@ public long maybeUpdate(long now) {
     }
 
     @Override
-    public void handleDisconnection(String destination) {
-        // Do nothing
-    }
-
-    @Override
-    public void handleFatalException(KafkaException exception) {
-        // We don't fail the broker on failures, but there should be sufficient information in the logs indicating the reason
-        // for failure.
-        log.debug("An error occurred in broker-to-broker communication.", exception);
+    public void handleServerDisconnect(long now, String nodeId, Optional<AuthenticationException> maybeAuthException) {
+        // We don't fail the broker on failures. There should be sufficient information from
+        // the NetworkClient logs to indicate the reason for the failure.
     }
 
     @Override
-    public void handleCompletedMetadataResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
+    public void handleFailedRequest(long now, Optional<KafkaException> maybeFatalException) {
         // Do nothing
     }
 
     @Override
-    public void requestUpdate() {
+    public void handleSuccessfulResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
         // Do nothing
     }
 

clients/src/main/java/org/apache/kafka/clients/Metadata.java

@@ -209,10 +209,8 @@ public synchronized Optional<MetadataCache.PartitionInfoAndEpoch> partitionInfoI
         }
     }
 
-    public synchronized void bootstrap(List<InetSocketAddress> addresses, long now) {
+    public synchronized void bootstrap(List<InetSocketAddress> addresses) {
         this.needUpdate = true;
-        this.lastRefreshMs = now;
-        this.lastSuccessfulRefreshMs = now;
         this.updateVersion += 1;
         this.cache = MetadataCache.bootstrap(addresses);
     }
@@ -419,9 +417,18 @@ private void clearRecoverableErrors() {
      * Record an attempt to update the metadata that failed. We need to keep track of this
      * to avoid retrying immediately.
      */
-    public synchronized void failedUpdate(long now, KafkaException fatalException) {
+    public synchronized void failedUpdate(long now) {
         this.lastRefreshMs = now;
-        this.fatalException = fatalException;
+    }
+
+    /**
+     * Propagate a fatal error which affects the ability to fetch metadata for the cluster.
+     * Two examples are authentication and unsupported version exceptions.
+     *
+     * @param exception The fatal exception
+     */
+    public synchronized void fatalError(KafkaException exception) {
+        this.fatalException = exception;
     }
 
     /**

clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java

@@ -18,11 +18,14 @@
 
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.requests.MetadataResponse;
 import org.apache.kafka.common.requests.RequestHeader;
 
 import java.io.Closeable;
 import java.util.List;
+import java.util.Optional;
 
 /**
  * The interface used by `NetworkClient` to request cluster metadata info to be updated and to retrieve the cluster nodes
@@ -46,7 +49,7 @@ public interface MetadataUpdater extends Closeable {
      * Starts a cluster metadata update if needed and possible. Returns the time until the metadata update (which would
      * be 0 if an update has been started as a result of this call).
      *
-     * If the implementation relies on `NetworkClient` to send requests, `handleCompletedMetadataResponse` will be
+     * If the implementation relies on `NetworkClient` to send requests, `handleSuccessfulResponse` will be
      * invoked after the metadata response is received.
      *
      * The semantics of `needed` and `possible` are implementation-dependent and may take into account a number of
@@ -55,34 +58,32 @@ public interface MetadataUpdater extends Closeable {
     long maybeUpdate(long now);
 
     /**
-     * Handle disconnections for metadata requests.
+     * Handle a server disconnect.
      *
      * This provides a mechanism for the `MetadataUpdater` implementation to use the NetworkClient instance for its own
      * requests with special handling for disconnections of such requests.
-     * @param destination
+     *
+     * @param now Current time in milliseconds
+     * @param nodeId The id of the node that disconnected
+     * @param maybeAuthException Optional authentication error
      */
-    void handleDisconnection(String destination);
+    void handleServerDisconnect(long now, String nodeId, Optional<AuthenticationException> maybeAuthException);
 
     /**
-     * Handle failure. Propagate the exception if awaiting metadata.
+     * Handle a metadata request failure.
      *
-     * @param fatalException exception corresponding to the failure
+     * @param now Current time in milliseconds
+     * @param maybeFatalException Optional fatal error (e.g. {@link UnsupportedVersionException})
      */
-    void handleFatalException(KafkaException fatalException);
+    void handleFailedRequest(long now, Optional<KafkaException> maybeFatalException);
 
     /**
      * Handle responses for metadata requests.
      *
      * This provides a mechanism for the `MetadataUpdater` implementation to use the NetworkClient instance for its own
      * requests with special handling for completed receives of such requests.
      */
-    void handleCompletedMetadataResponse(RequestHeader requestHeader, long now, MetadataResponse metadataResponse);
-
-    /**
-     * Schedules an update of the current cluster metadata info. A subsequent call to `maybeUpdate` would trigger the
-     * start of the update if possible (see `maybeUpdate` for more information).
-     */
-    void requestUpdate();
+    void handleSuccessfulResponse(RequestHeader requestHeader, long now, MetadataResponse metadataResponse);
 
     /**
      * Close this updater.