Merge branch '7.2.x' into 7.3.x by rayokota

core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaStore.java

@@ -26,6 +26,7 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.config.TopicConfig;
 import org.apache.kafka.common.errors.RecordTooLargeException;
@@ -143,6 +144,9 @@ public void init() throws StoreInitializationException {
         new KafkaStoreReaderThread<>(this.bootstrapBrokers, topic, groupId,
                                      this.storeUpdateHandler, serializer, this.localStore,
                                      this.producer, this.noopKey, this.initialized, this.config);
+    // checkpoint could be updated once the reader thread starts. This could result in a
+    // race condition where schemas after the checkpoint during startup would be double counted.
+    final Map<TopicPartition, Long> checkpoints = new HashMap<>(kafkaTopicReader.checkpoints());
     this.kafkaTopicReader.start();
 
     try {
@@ -156,7 +160,7 @@ public void init() throws StoreInitializationException {
       throw new StoreInitializationException("Illegal state while initializing store. Store "
                                              + "was already initialized");
     }
-    this.storeUpdateHandler.cacheInitialized(new HashMap<>(kafkaTopicReader.checkpoints()));
+    this.storeUpdateHandler.cacheInitialized(checkpoints);
     initLatch.countDown();
   }