Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mumrah merged 5 commits into from
Sep 12, 2022
Merged

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

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
cc18580
Don't make snapshots on broker after metadata errors
mumrah Sep 6, 2022
ad798ca
Add a test and try/catch for delta apply
mumrah Sep 7, 2022
0ac599f
PR feedback
mumrah Sep 8, 2022
e1dda63
Throw if we can't apply a delta
mumrah Sep 8, 2022
3883451
Clarify broker state after error
mumrah Sep 12, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 32 additions & 11 deletions core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ import org.apache.kafka.server.common.ApiMessageAndVersion
import org.apache.kafka.server.fault.FaultHandler
import org.apache.kafka.snapshot.SnapshotReader

import java.util.concurrent.atomic.AtomicBoolean

@jsancio jsancio Sep 8, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comment applies to this code:

        try {
          delta.replay(highestMetadataOffset, epoch, messageAndVersion.message())
        } catch {
          case e: Throwable => snapshotName match {
            case None => metadataLoadingFaultHandler.handleFault(
              s"Error replaying metadata log record at offset ${_highestOffset}", e)
            case Some(name) => metadataLoadingFaultHandler.handleFault(
              s"Error replaying record ${index} from snapshot ${name} at offset ${_highestOffset}", e)
          }
        }

I think this code attempts to read and replay the entire committed log. I wonder if this code should be more conservative if it encounters an error replaying a record and only read the current batch before updating the image.

Note that this code is used for both snapshots and log segments. For snapshots, the entire snapshot needs to be in one delta update.


object BrokerMetadataListener {
val MetadataBatchProcessingTimeUs = "MetadataBatchProcessingTimeUs"
Expand All @@ -41,8 +43,19 @@ class BrokerMetadataListener(
val maxBytesBetweenSnapshots: Long,
val snapshotter: Option[MetadataSnapshotter],
brokerMetrics: BrokerServerMetrics,
metadataLoadingFaultHandler: FaultHandler
_metadataLoadingFaultHandler: FaultHandler

@jsancio jsancio Sep 7, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is unrelated to this diff but while reviewing this code, I noticed that we don't handle or count errors when generating a MetadataImage. Are these errors fatal? If these errors are fatal shouldn't errors when applying the image and delta to the rest of the broker (log manager, replica manager, etc.) also be fatal? cc @niket-goel

@mumrah mumrah Sep 7, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The errors handled by this FaultHandler are non-fatal. AFAIK only the active controller will experience fatal metadata errors. For BrokerMetadataListener, we just log an ERROR and increment the metrics.

We call this fault handler in three cases in BrokerMetadataListener:

  • Replaying a record within a committed batch
  • Replaying a record within a batch from a snapshot
  • Finalizing a snapshot

@jsancio jsancio Sep 7, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Failures in this code seems to be fatal given the latest code:

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

Lines 323 to 330 in cc18580

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)

@mumrah mumrah Sep 7, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I see what you mean. It does stand to reason that an error when handling a record would lead to an error when building the MetadataImage itself. Seems like we could cover the delta.apply errors with the same non-fatal fault handler.

@cmccabe cmccabe Sep 7, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, the apply errors should be handled by metadataLoadingFaultHandler as well. If apply fails we obviously cannot apply the update at all, so we should have a catch block with something like throw metadataLoadingFaultHandler.handleFault("delta apply error", ...)

) extends RaftClient.Listener[ApiMessageAndVersion] with KafkaMetricsGroup {

private val metadataFaultOccurred = new AtomicBoolean(false)
private val metadataLoadingFaultHandler: FaultHandler = new FaultHandler() {
override def handleFault(failureMessage: String, cause: Throwable): RuntimeException = {
if (metadataFaultOccurred.compareAndSet(false, true)) {
error("Disabling metadata snapshots until this broker is restarted.")
}
_metadataLoadingFaultHandler.handleFault(failureMessage, cause)
}
Comment on lines +51 to +59

@jsancio jsancio Sep 7, 2022 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This abstraction feels strange. For example, how does the operator monitor that Kafka has an issue and it is not generating snapshots? I assume that they need to monitor the metric for BrokerSeverMetrics.metadataLoadErrorCount which is updated from KafkaRaftServer. The disabling of snapshotting happens in BrokerMetadataListener which doesn't know about this metric.

I think the solution should make this relation explicit and not have it hidden or implemented across multiple layers of abstraction.

@showuon showuon Sep 7, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree with @jsancio , or maybe we can just shutdown the broker with error, since the metadata cache in this broker is already wrong?

@mumrah mumrah Sep 7, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not really messing with the abstraction here, nor relying on any metrics. I've just wrapped the provided FaultHandler with some additional logic to set this new boolean. Since this fault handler is getting invoked for the metadata error handling cases, it seemed like a convenient choice.

Another way I considered was to add a method on FaultHandler to let a caller know if it had been invoked. Is this more like what you are suggesting?

or maybe we can just shutdown the broker with error

@showuon we currently just keep going in the face of metadata errors on the broker as a best effort to keep partition availability going. If we kill the broker on an error, we will likely see all the brokers crash. We are still working on a disaster recovery procedure, but killing the brokers would probably make that even harder. @cmccabe may have more context on this

@cmccabe cmccabe Sep 7, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please read https://cwiki.apache.org/confluence/display/KAFKA/KIP-859%3A+Add+Metadata+Log+Processing+Error+Related+Metrics for a description of how we're handling metadata errors.

Keep in mind that we're not trying to revisit that KIP here, just fixing a corner case (not creating snapshots after errors)

@showuon showuon Sep 8, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the info. I see. I have no more comments. Thank you

}

private val logContext = new LogContext(s"[BrokerMetadataListener id=$brokerId] ")
private val log = logContext.logger(classOf[BrokerMetadataListener])
logIdent = logContext.logPrefix()
Expand Down Expand Up @@ -147,7 +160,9 @@ class BrokerMetadataListener(

private def maybeStartSnapshot(): Unit = {
snapshotter.foreach { snapshotter =>
if (snapshotter.maybeStartSnapshot(_highestTimestamp, _delta.apply())) {
if (metadataFaultOccurred.get()) {
trace("Not starting metadata snapshot since we previously had an error")

@mumrah mumrah Sep 6, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I went with TRACE here to avoid spamming the logs with this message on each metadata commit

@cmccabe cmccabe Sep 6, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Honestly I think this should be WARN or ERROR to draw our attention to it.

@mumrah mumrah Sep 7, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My concern is that in an emergency scenario when things are going badly, we could lose important logs due to log rotation if we're spamming a bunch of the same error/warn messages.

@cmccabe cmccabe Sep 7, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah. That's fair.

I guess it's OK like it is for now. I am going to refactor this as part of the effort to unify Image handling between broker + controller.

} else if (snapshotter.maybeStartSnapshot(_highestTimestamp, _delta.apply())) {
_bytesSinceLastSnapshot = 0L
}
}
Expand Down Expand Up @@ -307,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()

@jsancio jsancio Sep 8, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note that it is possible for _delta to include a lot of batches maybe even the entire log. I wonder that if the broker encounters an error applying a delta we want to instead rewind, generate and apply a delta per record batch.

@mumrah mumrah Sep 8, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Rewinding and re-applying does sound useful for some kind of automatic error mitigation, but I think it will be a quite a bit of work. As it stands, I believe the broker can only process metadata going forward.

I can think of a degenerate case we have today where loadBatches is able to process all but one record, but delta.apply cannot complete and so we can't publish any new metadata. Like you mention, I think the only way to mitigate a situation like this would be to produce smaller deltas to reduce the blast radius of a bad record.

_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 {
Comment thread
mumrah marked this conversation as resolved.
case t: Throwable => metadataLoadingFaultHandler.handleFault(s"Error applying metadata delta $delta", t)
}
}

override def handleLeaderChange(leaderAndEpoch: LeaderAndEpoch): Unit = {
Expand Down
60 changes: 58 additions & 2 deletions core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}

Expand All @@ -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,
Expand All @@ -53,7 +54,7 @@ class BrokerMetadataListenerTest {
maxBytesBetweenSnapshots = maxBytesBetweenSnapshots,
snapshotter = snapshotter,
brokerMetrics = metrics,
metadataLoadingFaultHandler = metadataLoadingFaultHandler
_metadataLoadingFaultHandler = faultHandler
)
}

Expand Down Expand Up @@ -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 = {
Expand All @@ -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)
Expand Down Expand Up @@ -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],
Expand Down