KAFKA-13357: store producer IDs in broker snapshots

core/src/test/scala/unit/kafka/server/MetadataCacheTest.scala

@@ -64,7 +64,7 @@ object MetadataCacheTest {
         val partialImage = new MetadataImage(
           new RaftOffsetAndEpoch(100, 10),
           image.features(), ClusterImage.EMPTY,
-          image.topics(), image.configs(), image.clientQuotas())
+          image.topics(), image.configs(), image.clientQuotas(), image.producerIds())
         val delta = new MetadataDelta(partialImage)
 
         def toRecord(broker: UpdateMetadataBroker): RegisterBrokerRecord = {

core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala

@@ -25,6 +25,7 @@ import java.util.concurrent.atomic.AtomicReference
 import java.util.concurrent.{CountDownLatch, TimeUnit}
 import java.util.stream.IntStream
 import java.util.{Collections, Optional, Properties}
+
 import kafka.api._
 import kafka.cluster.{BrokerEndPoint, Partition}
 import kafka.log._
@@ -50,8 +51,8 @@ import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
 import org.apache.kafka.common.requests._
 import org.apache.kafka.common.security.auth.KafkaPrincipal
 import org.apache.kafka.common.utils.{Time, Utils}
-import org.apache.kafka.common.{IsolationLevel, Node, TopicPartition, TopicIdPartition, Uuid}
-import org.apache.kafka.image.{ClientQuotasImage, ClusterImageTest, ConfigurationsImage, FeaturesImage, MetadataImage, TopicsDelta, TopicsImage}
+import org.apache.kafka.common.{IsolationLevel, Node, TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.image.{ClientQuotasImage, ClusterImageTest, ConfigurationsImage, FeaturesImage, MetadataImage, ProducerIdsImage, TopicsDelta, TopicsImage}
 import org.apache.kafka.raft.{OffsetAndEpoch => RaftOffsetAndEpoch}
 import org.easymock.EasyMock
 import org.junit.jupiter.api.Assertions._
@@ -3489,7 +3490,8 @@ class ReplicaManagerTest {
       ClusterImageTest.IMAGE1,
       topicsImage,
       ConfigurationsImage.EMPTY,
-      ClientQuotasImage.EMPTY
+      ClientQuotasImage.EMPTY,
+      ProducerIdsImage.EMPTY
     )
   }
 

metadata/src/main/java/org/apache/kafka/image/MetadataDelta.java

@@ -25,6 +25,7 @@
 import org.apache.kafka.common.metadata.MetadataRecordType;
 import org.apache.kafka.common.metadata.PartitionChangeRecord;
 import org.apache.kafka.common.metadata.PartitionRecord;
+import org.apache.kafka.common.metadata.ProducerIdsRecord;
 import org.apache.kafka.common.metadata.RegisterBrokerRecord;
 import org.apache.kafka.common.metadata.RemoveFeatureLevelRecord;
 import org.apache.kafka.common.metadata.RemoveTopicRecord;
@@ -61,6 +62,8 @@ public final class MetadataDelta {
 
     private ClientQuotasDelta clientQuotasDelta = null;
 
+    private ProducerIdsDelta producerIdsDelta = null;
+
     public MetadataDelta(MetadataImage image) {
         this.image = image;
         this.highestOffset = image.highestOffsetAndEpoch().offset;
@@ -75,22 +78,58 @@ public FeaturesDelta featuresDelta() {
         return featuresDelta;
     }
 
+    public FeaturesDelta getOrCreateFeaturesDelta() {
+        if (featuresDelta == null) featuresDelta = new FeaturesDelta(image.features());
+        return featuresDelta;
+    }
+
     public ClusterDelta clusterDelta() {
         return clusterDelta;
     }
 
+    public ClusterDelta getOrCreateClusterDelta() {
+        if (clusterDelta == null) clusterDelta = new ClusterDelta(image.cluster());
+        return clusterDelta;
+    }
+
     public TopicsDelta topicsDelta() {
         return topicsDelta;
     }
 
+    public TopicsDelta getOrCreateTopicsDelta() {
+        if (topicsDelta == null) topicsDelta = new TopicsDelta(image.topics());
+        return topicsDelta;
+    }
+
     public ConfigurationsDelta configsDelta() {
         return configsDelta;
     }
 
+    public ConfigurationsDelta getOrCreateConfigsDelta() {
+        if (configsDelta == null) configsDelta = new ConfigurationsDelta(image.configs());
+        return configsDelta;
+    }
+
     public ClientQuotasDelta clientQuotasDelta() {
         return clientQuotasDelta;
     }
 
+    public ClientQuotasDelta getOrCreateClientQuotasDelta() {
+        if (clientQuotasDelta == null) clientQuotasDelta = new ClientQuotasDelta(image.clientQuotas());
+        return clientQuotasDelta;
+    }
+
+    public ProducerIdsDelta producerIdsDelta() {
+        return producerIdsDelta;
+    }
+
+    public ProducerIdsDelta getOrCreateProducerIdsDelta() {
+        if (producerIdsDelta == null) {
+            producerIdsDelta = new ProducerIdsDelta(image.producerIds());
+        }
+        return producerIdsDelta;
+    }
+
     public void read(long highestOffset, int highestEpoch, Iterator<List<ApiMessageAndVersion>> reader) {
         while (reader.hasNext()) {
             List<ApiMessageAndVersion> batch = reader.next();
@@ -140,7 +179,7 @@ public void replay(long offset, int epoch, ApiMessage record) {
                 replay((ClientQuotaRecord) record);
                 break;
             case PRODUCER_IDS_RECORD:
-                // Nothing to do.
+                replay((ProducerIdsRecord) record);
                 break;
             case REMOVE_FEATURE_LEVEL_RECORD:
                 replay((RemoveFeatureLevelRecord) record);
@@ -164,72 +203,66 @@ public void replay(UnregisterBrokerRecord record) {
     }
 
     public void replay(TopicRecord record) {
-        if (topicsDelta == null) topicsDelta = new TopicsDelta(image.topics());
-        topicsDelta.replay(record);
+        getOrCreateTopicsDelta().replay(record);
     }
 
     public void replay(PartitionRecord record) {
-        if (topicsDelta == null) topicsDelta = new TopicsDelta(image.topics());
-        topicsDelta.replay(record);
+        getOrCreateTopicsDelta().replay(record);
     }
 
     public void replay(ConfigRecord record) {
-        if (configsDelta == null) configsDelta = new ConfigurationsDelta(image.configs());
-        configsDelta.replay(record);
+        getOrCreateConfigsDelta().replay(record);
     }
 
     public void replay(PartitionChangeRecord record) {
-        if (topicsDelta == null) topicsDelta = new TopicsDelta(image.topics());
-        topicsDelta.replay(record);
+        getOrCreateTopicsDelta().replay(record);
     }
 
     public void replay(FenceBrokerRecord record) {
-        if (clusterDelta == null) clusterDelta = new ClusterDelta(image.cluster());
-        clusterDelta.replay(record);
+        getOrCreateClusterDelta().replay(record);
     }
 
     public void replay(UnfenceBrokerRecord record) {
-        if (clusterDelta == null) clusterDelta = new ClusterDelta(image.cluster());
-        clusterDelta.replay(record);
+        getOrCreateClusterDelta().replay(record);
     }
 
     public void replay(RemoveTopicRecord record) {
-        if (topicsDelta == null) topicsDelta = new TopicsDelta(image.topics());
+        getOrCreateTopicsDelta().replay(record);
         String topicName = topicsDelta.replay(record);
-        if (configsDelta == null) configsDelta = new ConfigurationsDelta(image.configs());
-        configsDelta.replay(record, topicName);
+        getOrCreateConfigsDelta().replay(record, topicName);
     }
 
     public void replay(FeatureLevelRecord record) {
-        if (featuresDelta == null) featuresDelta = new FeaturesDelta(image.features());
-        featuresDelta.replay(record);
+        getOrCreateFeaturesDelta().replay(record);
     }
 
     public void replay(BrokerRegistrationChangeRecord record) {
-        if (clusterDelta == null) clusterDelta = new ClusterDelta(image.cluster());
-        clusterDelta.replay(record);
+        getOrCreateClusterDelta().replay(record);
     }
 
     public void replay(ClientQuotaRecord record) {
-        if (clientQuotasDelta == null) clientQuotasDelta = new ClientQuotasDelta(image.clientQuotas());
-        clientQuotasDelta.replay(record);
+        getOrCreateClientQuotasDelta().replay(record);
+    }
+
+    public void replay(ProducerIdsRecord record) {
+        getOrCreateProducerIdsDelta().replay(record);
     }
 
     public void replay(RemoveFeatureLevelRecord record) {
-        if (featuresDelta == null) featuresDelta = new FeaturesDelta(image.features());
-        featuresDelta.replay(record);
+        getOrCreateFeaturesDelta().replay(record);
     }
 
     /**
      * Create removal deltas for anything which was in the base image, but which was not
      * referenced in the snapshot records we just applied.
      */
     public void finishSnapshot() {
-        if (featuresDelta != null) featuresDelta.finishSnapshot();
-        if (clusterDelta != null) clusterDelta.finishSnapshot();
-        if (topicsDelta != null) topicsDelta.finishSnapshot();
-        if (configsDelta != null) configsDelta.finishSnapshot();
-        if (clientQuotasDelta != null) clientQuotasDelta.finishSnapshot();
+        getOrCreateFeaturesDelta().finishSnapshot();
+        getOrCreateClusterDelta().finishSnapshot();
+        getOrCreateTopicsDelta().finishSnapshot();
+        getOrCreateConfigsDelta().finishSnapshot();
+        getOrCreateClientQuotasDelta().finishSnapshot();
+        getOrCreateProducerIdsDelta().finishSnapshot();
     }
 
     public MetadataImage apply() {
@@ -263,13 +296,20 @@ public MetadataImage apply() {
         } else {
             newClientQuotas = clientQuotasDelta.apply();
         }
+        ProducerIdsImage newProducerIds;
+        if (producerIdsDelta == null) {
+            newProducerIds = image.producerIds();
+        } else {
+            newProducerIds = producerIdsDelta.apply();
+        }
         return new MetadataImage(
             new OffsetAndEpoch(highestOffset, highestEpoch),
             newFeatures,
             newCluster,
             newTopics,
             newConfigs,
-            newClientQuotas
+            newClientQuotas,
+            newProducerIds
         );
     }
 
@@ -283,6 +323,7 @@ public String toString() {
             ", topicsDelta=" + topicsDelta +
             ", configsDelta=" + configsDelta +
             ", clientQuotasDelta=" + clientQuotasDelta +
+            ", producerIdsDelta=" + producerIdsDelta +
             ')';
     }
 }

metadata/src/main/java/org/apache/kafka/image/MetadataImage.java

@@ -37,7 +37,8 @@ public final class MetadataImage {
         ClusterImage.EMPTY,
         TopicsImage.EMPTY,
         ConfigurationsImage.EMPTY,
-        ClientQuotasImage.EMPTY);
+        ClientQuotasImage.EMPTY,
+        ProducerIdsImage.EMPTY);
 
     private final OffsetAndEpoch highestOffsetAndEpoch;
 
@@ -51,28 +52,33 @@ public final class MetadataImage {
 
     private final ClientQuotasImage clientQuotas;
 
+    private final ProducerIdsImage producerIds;
+
     public MetadataImage(
         OffsetAndEpoch highestOffsetAndEpoch,
         FeaturesImage features,
         ClusterImage cluster,
         TopicsImage topics,
         ConfigurationsImage configs,
-        ClientQuotasImage clientQuotas
+        ClientQuotasImage clientQuotas,
+        ProducerIdsImage producerIds
     ) {
         this.highestOffsetAndEpoch = highestOffsetAndEpoch;
         this.features = features;
         this.cluster = cluster;
         this.topics = topics;
         this.configs = configs;
         this.clientQuotas = clientQuotas;
+        this.producerIds = producerIds;
     }
 
     public boolean isEmpty() {
         return features.isEmpty() &&
             cluster.isEmpty() &&
             topics.isEmpty() &&
             configs.isEmpty() &&
-            clientQuotas.isEmpty();
+            clientQuotas.isEmpty() &&
+            producerIds.isEmpty();
     }
 
     public OffsetAndEpoch highestOffsetAndEpoch() {
@@ -99,12 +105,17 @@ public ClientQuotasImage clientQuotas() {
         return clientQuotas;
     }
 
+    public ProducerIdsImage producerIds() {
+        return producerIds;
+    }
+
     public void write(Consumer<List<ApiMessageAndVersion>> out) {
         features.write(out);
         cluster.write(out);
         topics.write(out);
         configs.write(out);
         clientQuotas.write(out);
+        producerIds.write(out);
     }
 
     @Override
@@ -116,12 +127,19 @@ public boolean equals(Object o) {
             cluster.equals(other.cluster) &&
             topics.equals(other.topics) &&
             configs.equals(other.configs) &&
-            clientQuotas.equals(other.clientQuotas);
+            clientQuotas.equals(other.clientQuotas) &&
+            producerIds.equals(other.producerIds);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(highestOffsetAndEpoch, features, cluster, topics, configs, clientQuotas);
+        return Objects.hash(highestOffsetAndEpoch,
+            features,
+            cluster,
+            topics,
+            configs,
+            clientQuotas,
+            producerIds);
     }
 
     @Override
@@ -132,6 +150,7 @@ public String toString() {
             ", topics=" + topics +
             ", configs=" + configs +
             ", clientQuotas=" + clientQuotas +
+            ", producerIdsImage=" + producerIds +
             ")";
     }
 }

metadata/src/main/java/org/apache/kafka/image/ProducerIdsDelta.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.image;
+
+import org.apache.kafka.common.metadata.ProducerIdsRecord;
+
+
+public final class ProducerIdsDelta {
+    private long highestSeenProducerId;
+
+    public ProducerIdsDelta(ProducerIdsImage image) {
+        this.highestSeenProducerId = image.highestSeenProducerId();
+    }
+
+    public void setHighestSeenProducerId(long highestSeenProducerId) {
+        this.highestSeenProducerId = highestSeenProducerId;
+    }
+
+    public long highestSeenProducerId() {
+        return highestSeenProducerId;
+    }
+
+    public void finishSnapshot() {
+        // Nothing to do
+    }
+
+    public void replay(ProducerIdsRecord record) {
+        highestSeenProducerId = record.producerIdsEnd();
+    }
+
+    public ProducerIdsImage apply() {
+        return new ProducerIdsImage(highestSeenProducerId);
+    }
+}