MINOR: Standardize controller log4j output for replaying records

metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java

@@ -26,6 +26,7 @@
 import org.apache.kafka.common.metadata.AccessControlEntryRecord;
 import org.apache.kafka.common.metadata.RemoveAccessControlEntryRecord;
 import org.apache.kafka.common.requests.ApiError;
+import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer;
 import org.apache.kafka.metadata.authorizer.StandardAcl;
 import org.apache.kafka.metadata.authorizer.StandardAclWithId;
@@ -37,6 +38,7 @@
 import org.apache.kafka.timeline.SnapshotRegistry;
 import org.apache.kafka.timeline.TimelineHashMap;
 import org.apache.kafka.timeline.TimelineHashSet;
+import org.slf4j.Logger;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -64,12 +66,44 @@
  * completed, which is another reason the prepare / complete callbacks are needed.
  */
 public class AclControlManager {
+    static class Builder {
+        private LogContext logContext = null;
+        private SnapshotRegistry snapshotRegistry = null;
+        private Optional<ClusterMetadataAuthorizer> authorizer = Optional.empty();
+
+        Builder setLogContext(LogContext logContext) {
+            this.logContext = logContext;
+            return this;
+        }
+
+        Builder setSnapshotRegistry(SnapshotRegistry snapshotRegistry) {
+            this.snapshotRegistry = snapshotRegistry;
+            return this;
+        }
+
+        Builder setClusterMetadataAuthorizer(Optional<ClusterMetadataAuthorizer> authorizer) {
+            this.authorizer = authorizer;
+            return this;
+        }
+
+        AclControlManager build() {
+            if (logContext == null) logContext = new LogContext();
+            if (snapshotRegistry == null) snapshotRegistry = new SnapshotRegistry(logContext);
+            return new AclControlManager(logContext, snapshotRegistry, authorizer);
+        }
+    }
+
+    private final Logger log;
     private final TimelineHashMap<Uuid, StandardAcl> idToAcl;
     private final TimelineHashSet<StandardAcl> existingAcls;
     private final Optional<ClusterMetadataAuthorizer> authorizer;
 
-    AclControlManager(SnapshotRegistry snapshotRegistry,
-                      Optional<ClusterMetadataAuthorizer> authorizer) {
+    AclControlManager(
+        LogContext logContext,
+        SnapshotRegistry snapshotRegistry,
+        Optional<ClusterMetadataAuthorizer> authorizer
+    ) {
+        this.log = logContext.logger(AclControlManager.class);
         this.idToAcl = new TimelineHashMap<>(snapshotRegistry, 0);
         this.existingAcls = new TimelineHashSet<>(snapshotRegistry, 0);
         this.authorizer = authorizer;
@@ -184,8 +218,10 @@ static void validateFilter(AclBindingFilter filter) {
         }
     }
 
-    public void replay(AccessControlEntryRecord record,
-                       Optional<OffsetAndEpoch> snapshotId) {
+    public void replay(
+        AccessControlEntryRecord record,
+        Optional<OffsetAndEpoch> snapshotId
+    ) {
         StandardAclWithId aclWithId = StandardAclWithId.fromRecord(record);
         idToAcl.put(aclWithId.id(), aclWithId.acl());
         existingAcls.add(aclWithId.acl());
@@ -194,10 +230,14 @@ public void replay(AccessControlEntryRecord record,
                 a.addAcl(aclWithId.id(), aclWithId.acl());
             });
         }
+        log.info("Replayed AccessControlEntryRecord for {}, setting {}", record.id(),
+                aclWithId.acl());
     }
 
-    public void replay(RemoveAccessControlEntryRecord record,
-                       Optional<OffsetAndEpoch> snapshotId) {
+    public void replay(
+        RemoveAccessControlEntryRecord record,
+        Optional<OffsetAndEpoch> snapshotId
+    ) {
         StandardAcl acl = idToAcl.remove(record.id());
         if (acl == null) {
             throw new RuntimeException("Unable to replay " + record + ": no acl with " +
@@ -212,6 +252,7 @@ public void replay(RemoveAccessControlEntryRecord record,
                 a.removeAcl(record.id());
             });
         }
+        log.info("Replayed RemoveAccessControlEntryRecord for {}, removing {}", record.id(), acl);
     }
 
     Map<Uuid, StandardAcl> idToAcl() {

metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java

@@ -26,9 +26,11 @@
 import org.apache.kafka.common.quota.ClientQuotaAlteration;
 import org.apache.kafka.common.quota.ClientQuotaEntity;
 import org.apache.kafka.common.requests.ApiError;
+import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
 import org.apache.kafka.timeline.SnapshotRegistry;
 import org.apache.kafka.timeline.TimelineHashMap;
+import org.slf4j.Logger;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
@@ -45,11 +47,38 @@
 
 
 public class ClientQuotaControlManager {
+    static class Builder {
+        private LogContext logContext = null;
+        private SnapshotRegistry snapshotRegistry = null;
+
+        Builder setLogContext(LogContext logContext) {
+            this.logContext = logContext;
+            return this;
+        }
+
+        Builder setSnapshotRegistry(SnapshotRegistry snapshotRegistry) {
+            this.snapshotRegistry = snapshotRegistry;
+            return this;
+        }
+
+        ClientQuotaControlManager build() {
+            if (logContext == null) logContext = new LogContext();
+            if (snapshotRegistry == null) snapshotRegistry = new SnapshotRegistry(logContext);
+            return new ClientQuotaControlManager(logContext, snapshotRegistry);
+        }
+    }
+
+    private final Logger log;
+
     private final SnapshotRegistry snapshotRegistry;
 
     final TimelineHashMap<ClientQuotaEntity, TimelineHashMap<String, Double>> clientQuotaData;
 
-    ClientQuotaControlManager(SnapshotRegistry snapshotRegistry) {
+    ClientQuotaControlManager(
+        LogContext logContext,
+        SnapshotRegistry snapshotRegistry
+    ) {
+        this.log = logContext.logger(ClientQuotaControlManager.class);
         this.snapshotRegistry = snapshotRegistry;
         this.clientQuotaData = new TimelineHashMap<>(snapshotRegistry, 0);
     }
@@ -109,8 +138,11 @@ public void replay(ClientQuotaRecord record) {
             if (quotas.size() == 0) {
                 clientQuotaData.remove(entity);
             }
+            log.info("Replayed ClientQuotaRecord for {} removing {}.", entity, record.key());
         } else {
             quotas.put(record.key(), record.value());
+            log.info("Replayed ClientQuotaRecord for {} setting {} to {}.",
+                    entity, record.key(), record.value());
         }
     }
 

metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java

@@ -437,11 +437,13 @@ public void replay(RegisterBrokerRecord record, long offset) {
             heartbeatManager.register(brokerId, record.fenced());
         }
         if (prevRegistration == null) {
-            log.info("Registered new broker: {}", record);
+            log.info("Replayed initial RegisterBrokerRecord for broker {}: {}", record.brokerId(), record);
         } else if (prevRegistration.incarnationId().equals(record.incarnationId())) {
-            log.info("Re-registered broker incarnation: {}", record);
+            log.info("Replayed RegisterBrokerRecord modifying the registration for broker {}: {}",
+                record.brokerId(), record);
         } else {
-            log.info("Re-registered broker id {}: {}", brokerId, record);
+            log.info("Replayed RegisterBrokerRecord establishing a new incarnation of broker {}: {}",
+                record.brokerId(), record);
         }
     }
 
@@ -458,7 +460,7 @@ public void replay(UnregisterBrokerRecord record) {
         } else {
             if (heartbeatManager != null) heartbeatManager.remove(brokerId);
             brokerRegistrations.remove(brokerId);
-            log.info("Unregistered broker: {}", record);
+            log.info("Replayed {}", record);
         }
     }
 
@@ -520,6 +522,8 @@ private void replayRegistrationChange(
                 inControlledShutdownChange
             );
             if (!curRegistration.equals(nextRegistration)) {
+                log.info("Replayed {} modifying the registration for broker {}: {}",
+                        record.getClass().getSimpleName(), brokerId, record);
                 brokerRegistrations.put(brokerId, nextRegistration);
             } else {
                 log.info("Ignoring no-op registration change for {}", curRegistration);

metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java

@@ -411,9 +411,11 @@ public void replay(ConfigRecord record) {
             configData.remove(configResource);
         }
         if (configSchema.isSensitive(record)) {
-            log.info("{}: set configuration {} to {}", configResource, record.name(), Password.HIDDEN);
+            log.info("Replayed ConfigRecord for {} which set configuration {} to {}",
+                    configResource, record.name(), Password.HIDDEN);
         } else {
-            log.info("{}: set configuration {} to {}", configResource, record.name(), record.value());
+            log.info("Replayed ConfigRecord for {} which set configuration {} to {}",
+                    configResource, record.name(), record.value());
         }
     }
 

metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java

@@ -325,24 +325,31 @@ public void replay(FeatureLevelRecord record) {
         }
         if (record.name().equals(MetadataVersion.FEATURE_NAME)) {
             MetadataVersion mv = MetadataVersion.fromFeatureLevel(record.featureLevel());
-            log.info("Setting metadata version to {}", mv);
             metadataVersion.set(mv);
+            log.info("Replayed a FeatureLevelRecord setting metadata version to {}", mv);
         } else {
             if (record.featureLevel() == 0) {
-                log.info("Removing feature {}", record.name());
                 finalizedVersions.remove(record.name());
+                log.info("Replayed a FeatureLevelRecord removing feature {}", record.name());
             } else {
-                log.info("Setting feature {} to {}", record.name(), record.featureLevel());
                 finalizedVersions.put(record.name(), record.featureLevel());
+                log.info("Replayed a FeatureLevelRecord setting feature {} to {}",
+                        record.name(), record.featureLevel());
             }
         }
     }
 
     public void replay(ZkMigrationStateRecord record) {
-        ZkMigrationState recordState = ZkMigrationState.of(record.zkMigrationState());
-        ZkMigrationState currentState = migrationControlState.get();
-        log.info("Transitioning ZK migration state from {} to {}", currentState, recordState);
-        migrationControlState.set(recordState);
+        ZkMigrationState newState = ZkMigrationState.of(record.zkMigrationState());
+        ZkMigrationState previousState = migrationControlState.get();
+        if (previousState.equals(newState)) {
+            log.debug("Replayed a ZkMigrationStateRecord which did not alter the state from {}.",
+                    previousState);
+        } else {
+            migrationControlState.set(newState);
+            log.info("Replayed a ZkMigrationStateRecord changing the migration state from {} to {}.",
+                    previousState, newState);
+        }
     }
 
     boolean isControllerId(int nodeId) {

metadata/src/main/java/org/apache/kafka/controller/ProducerIdControlManager.java

@@ -19,21 +19,62 @@
 
 import org.apache.kafka.common.errors.UnknownServerException;
 import org.apache.kafka.common.metadata.ProducerIdsRecord;
+import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
 import org.apache.kafka.server.common.ProducerIdsBlock;
 import org.apache.kafka.timeline.SnapshotRegistry;
 import org.apache.kafka.timeline.TimelineLong;
 import org.apache.kafka.timeline.TimelineObject;
+import org.slf4j.Logger;
 
 import java.util.Collections;
 
 
 public class ProducerIdControlManager {
+    static class Builder {
+        private LogContext logContext = null;
+        private SnapshotRegistry snapshotRegistry = null;
+        private ClusterControlManager clusterControlManager = null;
+
+        Builder setLogContext(LogContext logContext) {
+            this.logContext = logContext;
+            return this;
+        }
+
+        Builder setSnapshotRegistry(SnapshotRegistry snapshotRegistry) {
+            this.snapshotRegistry = snapshotRegistry;
+            return this;
+        }
+
+        Builder setClusterControlManager(ClusterControlManager clusterControlManager) {
+            this.clusterControlManager = clusterControlManager;
+            return this;
+        }
+
+        ProducerIdControlManager build() {
+            if (logContext == null) logContext = new LogContext();
+            if (snapshotRegistry == null) snapshotRegistry = new SnapshotRegistry(logContext);
+            if (clusterControlManager == null) {
+                throw new RuntimeException("You must specify ClusterControlManager.");
+            }
+            return new ProducerIdControlManager(
+                logContext,
+                clusterControlManager,
+                snapshotRegistry);
+        }
+    }
+
+    private final Logger log;
     private final ClusterControlManager clusterControlManager;
     private final TimelineObject<ProducerIdsBlock> nextProducerBlock;
     private final TimelineLong brokerEpoch;
 
-    ProducerIdControlManager(ClusterControlManager clusterControlManager, SnapshotRegistry snapshotRegistry) {
+    private ProducerIdControlManager(
+        LogContext logContext,
+        ClusterControlManager clusterControlManager,
+        SnapshotRegistry snapshotRegistry
+    ) {
+        this.log = logContext.logger(ProducerIdControlManager.class);
         this.clusterControlManager = clusterControlManager;
         this.nextProducerBlock = new TimelineObject<>(snapshotRegistry, ProducerIdsBlock.EMPTY);
         this.brokerEpoch = new TimelineLong(snapshotRegistry);
@@ -71,7 +112,9 @@ void replay(ProducerIdsRecord record) {
             throw new RuntimeException("Next Producer ID from replayed record (" + record.nextProducerId() + ")" +
                 " is not greater than current next Producer ID in block (" + nextBlock + ")");
         } else {
-            nextProducerBlock.set(new ProducerIdsBlock(record.brokerId(), record.nextProducerId(), ProducerIdsBlock.PRODUCER_ID_BLOCK_SIZE));
+            log.info("Replaying ProducerIdsRecord {}", record);
+            nextProducerBlock.set(new ProducerIdsBlock(record.brokerId(), record.nextProducerId(),
+                    ProducerIdsBlock.PRODUCER_ID_BLOCK_SIZE));
             brokerEpoch.set(record.brokerEpoch());
         }
     }

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -1829,7 +1829,10 @@ private QuorumController(
             setStaticConfig(staticConfig).
             setNodeId(nodeId).
             build();
-        this.clientQuotaControlManager = new ClientQuotaControlManager(snapshotRegistry);
+        this.clientQuotaControlManager = new ClientQuotaControlManager.Builder().
+            setLogContext(logContext).
+            setSnapshotRegistry(snapshotRegistry).
+            build();
         this.featureControl = new FeatureControlManager.Builder().
             setLogContext(logContext).
             setQuorumFeatures(quorumFeatures).
@@ -1851,7 +1854,11 @@ private QuorumController(
             setFeatureControlManager(featureControl).
             setZkMigrationEnabled(zkMigrationEnabled).
             build();
-        this.producerIdControlManager = new ProducerIdControlManager(clusterControl, snapshotRegistry);
+        this.producerIdControlManager = new ProducerIdControlManager.Builder().
+            setLogContext(logContext).
+            setSnapshotRegistry(snapshotRegistry).
+            setClusterControlManager(clusterControl).
+            build();
         this.leaderImbalanceCheckIntervalNs = leaderImbalanceCheckIntervalNs;
         this.maxIdleIntervalNs = maxIdleIntervalNs;
         this.replicationControl = new ReplicationControlManager.Builder().
@@ -1871,10 +1878,14 @@ private QuorumController(
             build();
         this.authorizer = authorizer;
         authorizer.ifPresent(a -> a.setAclMutator(this));
-        this.aclControlManager = new AclControlManager(snapshotRegistry, authorizer);
+        this.aclControlManager = new AclControlManager.Builder().
+            setLogContext(logContext).
+            setSnapshotRegistry(snapshotRegistry).
+            setClusterMetadataAuthorizer(authorizer).
+            build();
         this.logReplayTracker = new LogReplayTracker.Builder().
-                setLogContext(logContext).
-                build();
+            setLogContext(logContext).
+            build();
         this.raftClient = raftClient;
         this.bootstrapMetadata = bootstrapMetadata;
         this.maxRecordsPerBatch = maxRecordsPerBatch;