KAFKA-14203 Don't make snapshots on broker after metadata errors

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

@@ -322,16 +322,22 @@ class BrokerMetadataListener(
 
   private def publish(publisher: MetadataPublisher): Unit = {
     val delta = _delta
-    _image = _delta.apply()
-    _delta = new MetadataDelta(_image)
-    if (isDebugEnabled) {
-      debug(s"Publishing new metadata delta $delta at offset ${_image.highestOffsetAndEpoch().offset}.")
-    }
-    publisher.publish(delta, _image)
+    try {
+      _image = _delta.apply()
+      _delta = new MetadataDelta(_image)
+      if (isDebugEnabled) {
+        debug(s"Publishing new metadata delta $delta at offset ${_image.highestOffsetAndEpoch().offset}.")
+      }
 
-    // Update the metrics since the publisher handled the lastest image
-    brokerMetrics.lastAppliedRecordOffset.set(_highestOffset)
-    brokerMetrics.lastAppliedRecordTimestamp.set(_highestTimestamp)
+      // This publish call is done with its own try-catch and fault handler
+      publisher.publish(delta, _image)
+
+      // Update the metrics since the publisher handled the lastest image
+      brokerMetrics.lastAppliedRecordOffset.set(_highestOffset)
+      brokerMetrics.lastAppliedRecordTimestamp.set(_highestTimestamp)
+    } catch {
+      case t: Throwable => metadataLoadingFaultHandler.handleFault(s"Error applying metadata delta $delta", t)
+    }
   }
 
   override def handleLeaderChange(leaderAndEpoch: LeaderAndEpoch): Unit = {

core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala

@@ -27,7 +27,7 @@ import org.apache.kafka.common.{Endpoint, Uuid}
 import org.apache.kafka.image.{MetadataDelta, MetadataImage}
 import org.apache.kafka.metadata.{BrokerRegistration, RecordTestUtils, VersionRange}
 import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion}
-import org.apache.kafka.server.fault.MockFaultHandler
+import org.apache.kafka.server.fault.{FaultHandler, MockFaultHandler}
 import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
 import org.junit.jupiter.api.{AfterEach, Test}
 
@@ -45,6 +45,7 @@ class BrokerMetadataListenerTest {
     metrics: BrokerServerMetrics = BrokerServerMetrics(new Metrics()),
     snapshotter: Option[MetadataSnapshotter] = None,
     maxBytesBetweenSnapshots: Long = 1000000L,
+    faultHandler: FaultHandler = metadataLoadingFaultHandler
   ): BrokerMetadataListener = {
     new BrokerMetadataListener(
       brokerId = 0,
@@ -53,7 +54,7 @@ class BrokerMetadataListenerTest {
       maxBytesBetweenSnapshots = maxBytesBetweenSnapshots,
       snapshotter = snapshotter,
       brokerMetrics = metrics,
-      _metadataLoadingFaultHandler = metadataLoadingFaultHandler
+      _metadataLoadingFaultHandler = faultHandler
     )
   }
 
@@ -168,6 +169,7 @@ class BrokerMetadataListenerTest {
   }
 
   private val FOO_ID = Uuid.fromString("jj1G9utnTuCegi_gpnRgYw")
+  private val BAR_ID = Uuid.fromString("SzN5j0LvSEaRIJHrxfMAlg")
 
   private def generateManyRecords(listener: BrokerMetadataListener,
                                   endOffset: Long): Unit = {
@@ -192,6 +194,27 @@ class BrokerMetadataListenerTest {
     listener.getImageRecords().get()
   }
 
+  private def generateBadRecords(listener: BrokerMetadataListener,
+                                endOffset: Long): Unit = {
+    listener.handleCommit(
+      RecordTestUtils.mockBatchReader(
+        endOffset,
+        0,
+        util.Arrays.asList(
+          new ApiMessageAndVersion(new PartitionChangeRecord().
+            setPartitionId(0).
+            setTopicId(BAR_ID).
+            setRemovingReplicas(Collections.singletonList(1)), 0.toShort),
+          new ApiMessageAndVersion(new PartitionChangeRecord().
+            setPartitionId(0).
+            setTopicId(BAR_ID).
+            setRemovingReplicas(Collections.emptyList()), 0.toShort)
+        )
+      )
+    )
+    listener.getImageRecords().get()
+  }
+
   @Test
   def testHandleCommitsWithNoSnapshotterDefined(): Unit = {
     val listener = newBrokerMetadataListener(maxBytesBetweenSnapshots = 1000L)
@@ -289,6 +312,39 @@ class BrokerMetadataListenerTest {
     assertEquals(endOffset, snapshotter.activeSnapshotOffset, "We should generate snapshot on feature update")
   }
 
+  @Test
+  def testNoShapshotAfterError(): Unit = {
+    val snapshotter = new MockMetadataSnapshotter()
+    val faultHandler = new MockFaultHandler("metadata loading")
+
+    val listener = newBrokerMetadataListener(
+      snapshotter = Some(snapshotter),
+      maxBytesBetweenSnapshots = 1000L,
+      faultHandler = faultHandler)
+    try {
+      val brokerIds = 0 to 3
+
+      registerBrokers(listener, brokerIds, endOffset = 100L)
+      createTopicWithOnePartition(listener, replicas = brokerIds, endOffset = 200L)
+      listener.getImageRecords().get()
+      assertEquals(200L, listener.highestMetadataOffset)
+      assertEquals(-1L, snapshotter.prevCommittedOffset)
+      assertEquals(-1L, snapshotter.activeSnapshotOffset)
+
+      // Append invalid records that will normally trigger a snapshot
+      generateBadRecords(listener, 1000L)
+      assertEquals(-1L, snapshotter.prevCommittedOffset)
+      assertEquals(-1L, snapshotter.activeSnapshotOffset)
+
+      // Generate some records that will not throw an error, verify still no snapshots
+      generateManyRecords(listener, 2000L)
+      assertEquals(-1L, snapshotter.prevCommittedOffset)
+      assertEquals(-1L, snapshotter.activeSnapshotOffset)
+    } finally {
+      listener.close()
+    }
+  }
+
   private def registerBrokers(
     listener: BrokerMetadataListener,
     brokerIds: Iterable[Int],