MINOR: Refactor on FeaturesPublisher and ScramPublisher (apache#20522)

This PR is a follow-up from apache#20468.
Basically makes two things:
1. Moves the variable to the catch block as it is used only there.
2. Refactor FeaturesPublisher to handle exceptions the same as
ScramPublisher or other publishers :)

Reviewers: Chia-Ping Tsai <[email protected]>

---------

Signed-off-by: see-quick <[email protected]>

Author: see-quick

core/src/main/scala/kafka/server/ControllerServer.scala

@@ -145,7 +145,7 @@ class ControllerServer(
 
       metadataCachePublisher = new KRaftMetadataCachePublisher(metadataCache)
 
-      featuresPublisher = new FeaturesPublisher(logContext)
+      featuresPublisher = new FeaturesPublisher(logContext, sharedServer.metadataPublishingFaultHandler)
 
       registrationsPublisher = new ControllerRegistrationsPublisher()
 

core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala

@@ -224,7 +224,7 @@ class BrokerMetadataPublisher(
       dynamicTopicClusterQuotaPublisher.onMetadataUpdate(delta, newImage)
 
       // Apply SCRAM delta.
-      scramPublisher.onMetadataUpdate(delta, newImage)
+      scramPublisher.onMetadataUpdate(delta, newImage, manifest)
 
       // Apply DelegationToken delta.
       delegationTokenPublisher.onMetadataUpdate(delta, newImage)

metadata/src/main/java/org/apache/kafka/metadata/publisher/FeaturesPublisher.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.image.loader.LoaderManifest;
 import org.apache.kafka.image.publisher.MetadataPublisher;
 import org.apache.kafka.server.common.FinalizedFeatures;
+import org.apache.kafka.server.fault.FaultHandler;
 
 import org.slf4j.Logger;
 
@@ -31,12 +32,15 @@
 
 public class FeaturesPublisher implements MetadataPublisher {
     private final Logger log;
+    private final FaultHandler faultHandler;
     private volatile FinalizedFeatures finalizedFeatures = FinalizedFeatures.fromKRaftVersion(MINIMUM_VERSION);
 
     public FeaturesPublisher(
-        LogContext logContext
+        LogContext logContext,
+        FaultHandler faultHandler
     ) {
-        log = logContext.logger(FeaturesPublisher.class);
+        this.log = logContext.logger(FeaturesPublisher.class);
+        this.faultHandler = faultHandler;
     }
 
     public FinalizedFeatures features() {
@@ -54,15 +58,20 @@ public void onMetadataUpdate(
         MetadataImage newImage,
         LoaderManifest manifest
     ) {
-        if (delta.featuresDelta() != null) {
-            FinalizedFeatures newFinalizedFeatures = new FinalizedFeatures(newImage.features().metadataVersionOrThrow(),
+        try {
+            if (delta.featuresDelta() != null) {
+                FinalizedFeatures newFinalizedFeatures = new FinalizedFeatures(newImage.features().metadataVersionOrThrow(),
                     newImage.features().finalizedVersions(),
                     newImage.provenance().lastContainedOffset()
-            );
-            if (!newFinalizedFeatures.equals(finalizedFeatures)) {
-                log.info("Loaded new metadata {}.", newFinalizedFeatures);
-                finalizedFeatures = newFinalizedFeatures;
+                );
+                if (!newFinalizedFeatures.equals(finalizedFeatures)) {
+                    log.info("Loaded new metadata {}.", newFinalizedFeatures);
+                    finalizedFeatures = newFinalizedFeatures;
+                }
             }
+        } catch (Throwable t) {
+            faultHandler.handleFault("Uncaught exception while publishing SCRAM changes from MetadataDelta up to "
+                + newImage.highestOffsetAndEpoch().offset(), t);
         }
     }
 }

metadata/src/main/java/org/apache/kafka/metadata/publisher/ScramPublisher.java

@@ -44,11 +44,6 @@ public final String name() {
 
     @Override
     public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage, LoaderManifest manifest) {
-        onMetadataUpdate(delta, newImage);
-    }
-
-    public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage) {
-        String deltaName = "MetadataDelta up to " + newImage.highestOffsetAndEpoch().offset();
         try {
             // Apply changes to SCRAM credentials.
             ScramDelta scramDelta = delta.scramDelta();
@@ -63,7 +58,8 @@ public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage) {
                 });
             }
         } catch (Throwable t) {
-            faultHandler.handleFault("Uncaught exception while publishing SCRAM changes from " + deltaName, t);
+            faultHandler.handleFault("Uncaught exception while publishing SCRAM changes from MetadataDelta up to "
+                + newImage.highestOffsetAndEpoch().offset(), t);
         }
     }
 }