This is an automated email from the ASF dual-hosted git repository. cmccabe pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push: new 555744da70 KAFKA-14124: improve quorum controller fault handling (#12447) 555744da70 is described below commit 555744da7040f1ad91decc3cf2b813285af60aa2 Author: Colin Patrick McCabe <cmcc...@apache.org> AuthorDate: Thu Aug 4 22:49:45 2022 -0700 KAFKA-14124: improve quorum controller fault handling (#12447) Before trying to commit a batch of records to the __cluster_metadata log, the active controller should try to apply them to its current in-memory state. If this application process fails, the active controller process should exit, allowing another node to take leadership. This will prevent most bad metadata records from ending up in the log and help to surface errors during testing. Similarly, if the active controller attempts to renounce leadership, and the renunciation process itself fails, the process should exit. This will help avoid bugs where the active controller continues in an undefined state. In contrast, standby controllers that experience metadata application errors should continue on, in order to avoid a scenario where a bad record brings down the whole controller cluster. The intended effect of these changes is to make it harder to commit a bad record to the metadata log, but to continue to ride out the bad record as well as possible if such a record does get committed. This PR introduces the FaultHandler interface to implement these concepts. In junit tests, we use a FaultHandler implementation which does not exit the process. This allows us to avoid terminating the gradle test runner, which would be very disruptive. It also allows us to ensure that the test surfaces these exceptions, which we previously were not doing (the mock fault handler stores the exception). In addition to the above, this PR fixes a bug where RaftClient#resign was not being called from the renounce() function. This bug could have resulted in the raft layer not being informed of an active controller resigning. Reviewers: David Arthur <mum...@gmail.com> --- build.gradle | 2 + checkstyle/import-control-core.xml | 1 + checkstyle/import-control.xml | 4 + checkstyle/suppressions.xml | 2 + .../main/scala/kafka/server/ControllerServer.scala | 10 +- .../main/scala/kafka/server/KafkaRaftServer.scala | 6 +- .../java/kafka/testkit/KafkaClusterTestKit.java | 36 +- .../kafka/server/QuorumTestHarness.scala | 6 + .../apache/kafka/controller/QuorumController.java | 382 ++++++++++++--------- .../metadata/fault/MetadataFaultException.java | 32 ++ .../kafka/metadata/fault/MetadataFaultHandler.java | 36 ++ .../kafka/controller/QuorumControllerTest.java | 25 ++ .../kafka/controller/QuorumControllerTestEnv.java | 15 + .../apache/kafka/server/fault/FaultHandler.java | 58 ++++ .../server/fault/ProcessExitingFaultHandler.java | 37 ++ .../kafka/server/fault/MockFaultHandler.java | 65 ++++ .../server/fault/MockFaultHandlerException.java | 38 ++ 17 files changed, 586 insertions(+), 169 deletions(-) diff --git a/build.gradle b/build.gradle index 54068f2977..e1f773075a 100644 --- a/build.gradle +++ b/build.gradle @@ -897,6 +897,7 @@ project(':core') { testImplementation project(':clients').sourceSets.test.output testImplementation project(':metadata').sourceSets.test.output testImplementation project(':raft').sourceSets.test.output + testImplementation project(':server-common').sourceSets.test.output testImplementation libs.bcpkix testImplementation libs.mockitoCore testImplementation(libs.apacheda) { @@ -1179,6 +1180,7 @@ project(':metadata') { testImplementation libs.slf4jlog4j testImplementation project(':clients').sourceSets.test.output testImplementation project(':raft').sourceSets.test.output + testImplementation project(':server-common').sourceSets.test.output generator project(':generator') } diff --git a/checkstyle/import-control-core.xml b/checkstyle/import-control-core.xml index 28b325b093..4042cba402 100644 --- a/checkstyle/import-control-core.xml +++ b/checkstyle/import-control-core.xml @@ -54,6 +54,7 @@ <allow pkg="org.apache.kafka.metadata" /> <allow pkg="org.apache.kafka.metalog" /> <allow pkg="org.apache.kafka.server.common" /> + <allow pkg="org.apache.kafka.server.fault" /> </subpackage> <subpackage name="tools"> diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml index 211d23ff60..4b07a26cba 100644 --- a/checkstyle/import-control.xml +++ b/checkstyle/import-control.xml @@ -232,6 +232,7 @@ <allow pkg="org.apache.kafka.raft" /> <allow pkg="org.apache.kafka.server.authorizer" /> <allow pkg="org.apache.kafka.server.common" /> + <allow pkg="org.apache.kafka.server.fault" /> <allow pkg="org.apache.kafka.server.metrics" /> <allow pkg="org.apache.kafka.server.policy"/> <allow pkg="org.apache.kafka.snapshot" /> @@ -276,6 +277,9 @@ <allow pkg="org.apache.kafka.controller" /> <allow pkg="org.apache.kafka.metadata" /> </subpackage> + <subpackage name="fault"> + <allow pkg="org.apache.kafka.server.fault" /> + </subpackage> </subpackage> <subpackage name="metalog"> diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index f6ca0d02fe..bec3da1637 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -39,6 +39,8 @@ <suppress checks="(NPathComplexity|ClassFanOutComplexity|CyclomaticComplexity|ClassDataAbstractionCoupling|FinalLocalVariable|LocalVariableName|MemberName|ParameterName|MethodLength|JavaNCSS|AvoidStarImport)" files="core[\\/]src[\\/](generated|generated-test)[\\/].+.java$"/> <suppress checks="NPathComplexity" files="(ClusterTestExtensions|KafkaApisBuilder).java"/> + <suppress checks="MethodLength" + files="(KafkaClusterTestKit).java"/> <!-- Clients --> <suppress id="dontUseSystemExit" diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala b/core/src/main/scala/kafka/server/ControllerServer.scala index cff88d2b6b..212c092e1a 100644 --- a/core/src/main/scala/kafka/server/ControllerServer.scala +++ b/core/src/main/scala/kafka/server/ControllerServer.scala @@ -21,7 +21,6 @@ import java.util import java.util.OptionalLong import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.{CompletableFuture, TimeUnit} - import kafka.cluster.Broker.ServerInfo import kafka.metrics.{KafkaMetricsGroup, LinuxIoMetricsCollector} import kafka.network.{DataPlaneAcceptor, SocketServer} @@ -46,6 +45,7 @@ import org.apache.kafka.server.authorizer.Authorizer import org.apache.kafka.server.common.ApiMessageAndVersion import org.apache.kafka.common.config.ConfigException import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer +import org.apache.kafka.server.fault.FaultHandler import org.apache.kafka.server.metrics.KafkaYammerMetrics import org.apache.kafka.server.policy.{AlterConfigPolicy, CreateTopicPolicy} @@ -65,7 +65,9 @@ class ControllerServer( val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer, AddressSpec]], val configSchema: KafkaConfigSchema, val raftApiVersions: ApiVersions, - val bootstrapMetadata: BootstrapMetadata + val bootstrapMetadata: BootstrapMetadata, + val metadataFaultHandler: FaultHandler, + val fatalFaultHandler: FaultHandler, ) extends Logging with KafkaMetricsGroup { import kafka.server.Server._ @@ -204,7 +206,9 @@ class ControllerServer( setAlterConfigPolicy(alterConfigPolicy.asJava). setConfigurationValidator(new ControllerConfigurationValidator()). setStaticConfig(config.originals). - setBootstrapMetadata(bootstrapMetadata) + setBootstrapMetadata(bootstrapMetadata). + setMetadataFaultHandler(metadataFaultHandler). + setFatalFaultHandler(fatalFaultHandler) } authorizer match { case Some(a: ClusterMetadataAuthorizer) => controllerBuilder.setAuthorizer(a) diff --git a/core/src/main/scala/kafka/server/KafkaRaftServer.scala b/core/src/main/scala/kafka/server/KafkaRaftServer.scala index 07a3118372..e7cf8f8f1f 100644 --- a/core/src/main/scala/kafka/server/KafkaRaftServer.scala +++ b/core/src/main/scala/kafka/server/KafkaRaftServer.scala @@ -29,9 +29,11 @@ import org.apache.kafka.common.internals.Topic import org.apache.kafka.common.utils.{AppInfoParser, Time} import org.apache.kafka.common.{KafkaException, Uuid} import org.apache.kafka.controller.BootstrapMetadata +import org.apache.kafka.metadata.fault.MetadataFaultHandler import org.apache.kafka.metadata.{KafkaConfigSchema, MetadataRecordSerde} import org.apache.kafka.raft.RaftConfig import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion} +import org.apache.kafka.server.fault.ProcessExitingFaultHandler import org.apache.kafka.server.metrics.KafkaYammerMetrics import java.nio.file.Paths @@ -106,7 +108,9 @@ class KafkaRaftServer( controllerQuorumVotersFuture, KafkaRaftServer.configSchema, raftManager.apiVersions, - bootstrapMetadata + bootstrapMetadata, + new MetadataFaultHandler(), + new ProcessExitingFaultHandler(), )) } else { None diff --git a/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java b/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java index c961d71bbe..42120324f5 100644 --- a/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java +++ b/core/src/test/java/kafka/testkit/KafkaClusterTestKit.java @@ -40,6 +40,7 @@ import org.apache.kafka.metadata.MetadataRecordSerde; import org.apache.kafka.raft.RaftConfig; import org.apache.kafka.server.common.ApiMessageAndVersion; import org.apache.kafka.server.common.MetadataVersion; +import org.apache.kafka.server.fault.MockFaultHandler; import org.apache.kafka.test.TestUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -115,6 +116,8 @@ public class KafkaClusterTestKit implements AutoCloseable { public static class Builder { private TestKitNodes nodes; private Map<String, String> configProps = new HashMap<>(); + private MockFaultHandler metadataFaultHandler = new MockFaultHandler("metadataFaultHandler"); + private MockFaultHandler fatalFaultHandler = new MockFaultHandler("fatalFaultHandler"); public Builder(TestKitNodes nodes) { this.nodes = nodes; @@ -190,7 +193,9 @@ public class KafkaClusterTestKit implements AutoCloseable { connectFutureManager.future, KafkaRaftServer.configSchema(), raftManager.apiVersions(), - bootstrapMetadata + bootstrapMetadata, + metadataFaultHandler, + fatalFaultHandler ); controllers.put(node.id(), controller); controller.socketServerFirstBoundPortFuture().whenComplete((port, e) -> { @@ -273,7 +278,8 @@ public class KafkaClusterTestKit implements AutoCloseable { throw e; } return new KafkaClusterTestKit(executorService, nodes, controllers, - brokers, raftManagers, connectFutureManager, baseDirectory); + brokers, raftManagers, connectFutureManager, baseDirectory, + metadataFaultHandler, fatalFaultHandler); } private String listeners(int node) { @@ -314,14 +320,20 @@ public class KafkaClusterTestKit implements AutoCloseable { private final Map<Integer, KafkaRaftManager<ApiMessageAndVersion>> raftManagers; private final ControllerQuorumVotersFutureManager controllerQuorumVotersFutureManager; private final File baseDirectory; - - private KafkaClusterTestKit(ExecutorService executorService, - TestKitNodes nodes, - Map<Integer, ControllerServer> controllers, - Map<Integer, BrokerServer> brokers, - Map<Integer, KafkaRaftManager<ApiMessageAndVersion>> raftManagers, - ControllerQuorumVotersFutureManager controllerQuorumVotersFutureManager, - File baseDirectory) { + private final MockFaultHandler metadataFaultHandler; + private final MockFaultHandler fatalFaultHandler; + + private KafkaClusterTestKit( + ExecutorService executorService, + TestKitNodes nodes, + Map<Integer, ControllerServer> controllers, + Map<Integer, BrokerServer> brokers, + Map<Integer, KafkaRaftManager<ApiMessageAndVersion>> raftManagers, + ControllerQuorumVotersFutureManager controllerQuorumVotersFutureManager, + File baseDirectory, + MockFaultHandler metadataFaultHandler, + MockFaultHandler fatalFaultHandler + ) { this.executorService = executorService; this.nodes = nodes; this.controllers = controllers; @@ -329,6 +341,8 @@ public class KafkaClusterTestKit implements AutoCloseable { this.raftManagers = raftManagers; this.controllerQuorumVotersFutureManager = controllerQuorumVotersFutureManager; this.baseDirectory = baseDirectory; + this.metadataFaultHandler = metadataFaultHandler; + this.fatalFaultHandler = fatalFaultHandler; } public void format() throws Exception { @@ -520,6 +534,8 @@ public class KafkaClusterTestKit implements AutoCloseable { executorService.shutdownNow(); executorService.awaitTermination(5, TimeUnit.MINUTES); } + metadataFaultHandler.maybeRethrowFirstException(); + fatalFaultHandler.maybeRethrowFirstException(); } private void waitForAllFutures(List<Entry<String, Future<?>>> futureEntries) diff --git a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala index a2393cdccb..9894df9c5f 100755 --- a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala +++ b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala @@ -36,6 +36,7 @@ import org.apache.kafka.controller.BootstrapMetadata import org.apache.kafka.metadata.MetadataRecordSerde import org.apache.kafka.raft.RaftConfig.{AddressSpec, InetAddressSpec} import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion} +import org.apache.kafka.server.fault.MockFaultHandler import org.apache.zookeeper.client.ZKClientConfig import org.apache.zookeeper.{WatchedEvent, Watcher, ZooKeeper} import org.junit.jupiter.api.Assertions._ @@ -188,6 +189,8 @@ abstract class QuorumTestHarness extends Logging { } } + val faultHandler = new MockFaultHandler("quorumTestHarnessFaultHandler") + // Note: according to the junit documentation: "JUnit Jupiter does not guarantee the execution // order of multiple @BeforeEach methods that are declared within a single test class or test // interface." Therefore, if you have things you would like to do before each test case runs, it @@ -308,6 +311,8 @@ abstract class QuorumTestHarness extends Logging { configSchema = KafkaRaftServer.configSchema, raftApiVersions = raftManager.apiVersions, bootstrapMetadata = BootstrapMetadata.create(metadataVersion, bootstrapRecords.asJava), + metadataFaultHandler = faultHandler, + fatalFaultHandler = faultHandler, ) controllerServer.socketServerFirstBoundPortFuture.whenComplete((port, e) => { if (e != null) { @@ -374,6 +379,7 @@ abstract class QuorumTestHarness extends Logging { } System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM) Configuration.setConfiguration(null) + faultHandler.maybeRethrowFirstException() } // Trigger session expiry by reusing the session id in another client diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java index 0290e0040c..a4cc1d92cb 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -91,6 +91,7 @@ import org.apache.kafka.server.authorizer.AclCreateResult; import org.apache.kafka.server.authorizer.AclDeleteResult; import org.apache.kafka.server.common.ApiMessageAndVersion; import org.apache.kafka.server.common.MetadataVersion; +import org.apache.kafka.server.fault.FaultHandler; import org.apache.kafka.server.policy.AlterConfigPolicy; import org.apache.kafka.server.policy.CreateTopicPolicy; import org.apache.kafka.snapshot.SnapshotReader; @@ -149,6 +150,8 @@ public final class QuorumController implements Controller { static public class Builder { private final int nodeId; private final String clusterId; + private FaultHandler fatalFaultHandler = null; + private FaultHandler metadataFaultHandler = null; private Time time = Time.SYSTEM; private String threadNamePrefix = null; private LogContext logContext = null; @@ -175,6 +178,16 @@ public final class QuorumController implements Controller { this.clusterId = clusterId; } + public Builder setFatalFaultHandler(FaultHandler fatalFaultHandler) { + this.fatalFaultHandler = fatalFaultHandler; + return this; + } + + public Builder setMetadataFaultHandler(FaultHandler metadataFaultHandler) { + this.metadataFaultHandler = metadataFaultHandler; + return this; + } + public int nodeId() { return nodeId; } @@ -287,6 +300,10 @@ public final class QuorumController implements Controller { throw new IllegalStateException("You must specify an initial metadata.version using the kafka-storage tool."); } else if (quorumFeatures == null) { throw new IllegalStateException("You must specify the quorum features"); + } else if (fatalFaultHandler == null) { + throw new IllegalStateException("You must specify a fatal fault handler."); + } else if (metadataFaultHandler == null) { + throw new IllegalStateException("You must specify a metadata fault handler."); } if (threadNamePrefix == null) { @@ -304,6 +321,8 @@ public final class QuorumController implements Controller { try { queue = new KafkaEventQueue(time, logContext, threadNamePrefix + "QuorumController"); return new QuorumController( + fatalFaultHandler, + metadataFaultHandler, logContext, nodeId, clusterId, @@ -426,12 +445,18 @@ public final class QuorumController implements Controller { exception.getClass().getSimpleName(), deltaUs); return exception; } - log.warn("{}: failed with unknown server exception {} at epoch {} in {} us. " + - "Renouncing leadership and reverting to the last committed offset {}.", - name, exception.getClass().getSimpleName(), curClaimEpoch, deltaUs, - lastCommittedOffset, exception); - raftClient.resign(curClaimEpoch); - renounce(); + if (isActiveController()) { + log.warn("{}: failed with unknown server exception {} at epoch {} in {} us. " + + "Renouncing leadership and reverting to the last committed offset {}.", + name, exception.getClass().getSimpleName(), curClaimEpoch, deltaUs, + lastCommittedOffset, exception); + renounce(); + } else { + log.warn("{}: failed with unknown server exception {} in {} us. " + + "The controller is already in standby mode.", + name, exception.getClass().getSimpleName(), deltaUs, + exception); + } return new UnknownServerException(exception); } @@ -702,7 +727,7 @@ public final class QuorumController implements Controller { long now = time.nanoseconds(); controllerMetrics.updateEventQueueTime(NANOSECONDS.toMillis(now - eventCreatedTimeNs)); int controllerEpoch = curClaimEpoch; - if (controllerEpoch == -1) { + if (!isActiveController()) { throw newNotControllerException(); } startProcessingTimeNs = OptionalLong.of(now); @@ -728,9 +753,26 @@ public final class QuorumController implements Controller { "reaches offset {}", this, resultAndOffset.offset()); } } else { - // If the operation returned a batch of records, those records need to be - // written before we can return our result to the user. Here, we hand off - // the batch of records to the raft client. They will be written out + // Start by trying to apply the record to our in-memory state. This should always + // succeed; if it does not, that's a fatal error. It is important to do this before + // scheduling the record for Raft replication. + int i = 1; + for (ApiMessageAndVersion message : result.records()) { + try { + replay(message.message(), Optional.empty()); + } catch (Throwable e) { + String failureMessage = String.format("Unable to apply %s record, which was " + + "%d of %d record(s) in the batch following last writeOffset %d.", + message.message().getClass().getSimpleName(), i, result.records().size(), + writeOffset); + fatalFaultHandler.handleFault(failureMessage, e); + } + i++; + } + + // If the operation returned a batch of records, and those records could be applied, + // they need to be written before we can return our result to the user. Here, we + // hand off the batch of records to the raft client. They will be written out // asynchronously. final long offset; if (result.isAtomic()) { @@ -741,9 +783,6 @@ public final class QuorumController implements Controller { op.processBatchEndOffset(offset); updateWriteOffset(offset); resultAndOffset = ControllerResultAndOffset.of(offset, result); - for (ApiMessageAndVersion message : result.records()) { - replay(message.message(), Optional.empty(), offset); - } snapshotRegistry.getOrCreateSnapshot(offset); log.debug("Read-write operation {} will be completed when the log " + @@ -789,9 +828,9 @@ public final class QuorumController implements Controller { return event.future(); } - private <T> CompletableFuture<T> appendWriteEvent(String name, - OptionalLong deadlineNs, - ControllerWriteOperation<T> op) { + <T> CompletableFuture<T> appendWriteEvent(String name, + OptionalLong deadlineNs, + ControllerWriteOperation<T> op) { ControllerWriteEvent<T> event = new ControllerWriteEvent<>(name, op); if (deadlineNs.isPresent()) { queue.appendWithDeadline(deadlineNs.getAsLong(), event); @@ -841,11 +880,20 @@ public final class QuorumController implements Controller { "offset {} and epoch {}.", offset, epoch); } } - for (ApiMessageAndVersion messageAndVersion : messages) { - replay(messageAndVersion.message(), Optional.empty(), offset); + int i = 1; + for (ApiMessageAndVersion message : messages) { + try { + replay(message.message(), Optional.empty()); + } catch (Throwable e) { + String failureMessage = String.format("Unable to apply %s record on standby " + + "controller, which was %d of %d record(s) in the batch with baseOffset %d.", + message.message().getClass().getSimpleName(), i, messages.size(), + batch.baseOffset()); + metadataFaultHandler.handleFault(failureMessage, e); + } + i++; } } - updateLastCommittedState(offset, epoch, batch.appendTimestamp()); processedRecordsSize += batch.sizeInBytes(); } @@ -862,13 +910,9 @@ public final class QuorumController implements Controller { appendRaftEvent(String.format("handleSnapshot[snapshotId=%s]", reader.snapshotId()), () -> { try { if (isActiveController()) { - throw new IllegalStateException( - String.format( - "Asked to load snapshot (%s) when it is the active controller (%d)", - reader.snapshotId(), - curClaimEpoch - ) - ); + fatalFaultHandler.handleFault(String.format("Asked to load snapshot " + + "(%s) when it is the active controller (%d)", reader.snapshotId(), + curClaimEpoch)); } log.info("Starting to replay snapshot ({}), from last commit offset ({}) and epoch ({})", reader.snapshotId(), lastCommittedOffset, lastCommittedEpoch); @@ -882,26 +926,28 @@ public final class QuorumController implements Controller { if (log.isDebugEnabled()) { if (log.isTraceEnabled()) { - log.trace( - "Replaying snapshot ({}) batch with last offset of {}: {}", - reader.snapshotId(), - offset, - messages - .stream() - .map(ApiMessageAndVersion::toString) - .collect(Collectors.joining(", ")) - ); + log.trace("Replaying snapshot ({}) batch with last offset of {}: {}", + reader.snapshotId(), offset, messages.stream().map(ApiMessageAndVersion::toString). + collect(Collectors.joining(", "))); } else { - log.debug( - "Replaying snapshot ({}) batch with last offset of {}", - reader.snapshotId(), - offset - ); + log.debug("Replaying snapshot ({}) batch with last offset of {}", + reader.snapshotId(), offset); } } - for (ApiMessageAndVersion messageAndVersion : messages) { - replay(messageAndVersion.message(), Optional.of(reader.snapshotId()), offset); + int i = 1; + for (ApiMessageAndVersion message : messages) { + try { + replay(message.message(), Optional.of(reader.snapshotId())); + } catch (Throwable e) { + String failureMessage = String.format("Unable to apply %s record " + + "from snapshot %s on standby controller, which was %d of " + + "%d record(s) in the batch with baseOffset %d.", + message.message().getClass().getSimpleName(), reader.snapshotId(), + i, messages.size(), batch.baseOffset()); + metadataFaultHandler.handleFault(failureMessage, e); + } + i++; } } updateLastCommittedState( @@ -968,10 +1014,14 @@ public final class QuorumController implements Controller { if (exception != null) { log.error("Failed to bootstrap metadata.", exception); appendRaftEvent("bootstrapMetadata[" + curClaimEpoch + "]", () -> { - log.warn("Renouncing the leadership at oldEpoch {} since we could not bootstrap " + - "metadata. Reverting to last committed offset {}.", - curClaimEpoch, lastCommittedOffset); - renounce(); + if (isActiveController()) { + log.warn("Renouncing the leadership at oldEpoch {} since we could not bootstrap " + + "metadata. Reverting to last committed offset {}.", + curClaimEpoch, lastCommittedOffset); + renounce(); + } else { + log.warn("Unable to bootstrap metadata on standby controller."); + } }); } }); @@ -998,10 +1048,12 @@ public final class QuorumController implements Controller { }); } else if (isActiveController()) { appendRaftEvent("handleRenounce[" + curClaimEpoch + "]", () -> { - log.warn("Renouncing the leadership at oldEpoch {} due to a metadata " + - "log event. Reverting to last committed offset {}.", curClaimEpoch, - lastCommittedOffset); - renounce(); + if (isActiveController()) { + log.warn("Renouncing the leadership at oldEpoch {} due to a metadata " + + "log event. Reverting to last committed offset {}.", curClaimEpoch, + lastCommittedOffset); + renounce(); + } }); } } @@ -1078,26 +1130,34 @@ public final class QuorumController implements Controller { } private void renounce() { - curClaimEpoch = -1; - controllerMetrics.setActive(false); - purgatory.failAll(newNotControllerException()); - - if (snapshotRegistry.hasSnapshot(lastCommittedOffset)) { - newBytesSinceLastSnapshot = 0; - snapshotRegistry.revertToSnapshot(lastCommittedOffset); - authorizer.ifPresent(a -> a.loadSnapshot(aclControlManager.idToAcl())); - } else { - resetState(); - raftClient.unregister(metaLogListener); - metaLogListener = new QuorumMetaLogListener(); - raftClient.register(metaLogListener); + try { + if (curClaimEpoch == -1) { + throw new RuntimeException("Cannot renounce leadership because we are not the " + + "current leader."); + } + raftClient.resign(curClaimEpoch); + curClaimEpoch = -1; + controllerMetrics.setActive(false); + purgatory.failAll(newNotControllerException()); + + if (snapshotRegistry.hasSnapshot(lastCommittedOffset)) { + newBytesSinceLastSnapshot = 0; + snapshotRegistry.revertToSnapshot(lastCommittedOffset); + authorizer.ifPresent(a -> a.loadSnapshot(aclControlManager.idToAcl())); + } else { + resetState(); + raftClient.unregister(metaLogListener); + metaLogListener = new QuorumMetaLogListener(); + raftClient.register(metaLogListener); + } + updateWriteOffset(-1); + clusterControl.deactivate(); + cancelMaybeFenceReplicas(); + cancelMaybeBalancePartitionLeaders(); + cancelNextWriteNoOpRecord(); + } catch (Throwable e) { + fatalFaultHandler.handleFault("exception while renouncing leadership", e); } - - updateWriteOffset(-1); - clusterControl.deactivate(); - cancelMaybeFenceReplicas(); - cancelMaybeBalancePartitionLeaders(); - cancelNextWriteNoOpRecord(); } private <T> void scheduleDeferredWriteEvent(String name, long deadlineNs, @@ -1246,70 +1306,60 @@ public final class QuorumController implements Controller { } @SuppressWarnings("unchecked") - private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, long offset) { - try { - MetadataRecordType type = MetadataRecordType.fromId(message.apiKey()); - switch (type) { - case REGISTER_BROKER_RECORD: - clusterControl.replay((RegisterBrokerRecord) message); - break; - case UNREGISTER_BROKER_RECORD: - clusterControl.replay((UnregisterBrokerRecord) message); - break; - case TOPIC_RECORD: - replicationControl.replay((TopicRecord) message); - break; - case PARTITION_RECORD: - replicationControl.replay((PartitionRecord) message); - break; - case CONFIG_RECORD: - configurationControl.replay((ConfigRecord) message); - break; - case PARTITION_CHANGE_RECORD: - replicationControl.replay((PartitionChangeRecord) message); - break; - case FENCE_BROKER_RECORD: - clusterControl.replay((FenceBrokerRecord) message); - break; - case UNFENCE_BROKER_RECORD: - clusterControl.replay((UnfenceBrokerRecord) message); - break; - case REMOVE_TOPIC_RECORD: - replicationControl.replay((RemoveTopicRecord) message); - break; - case FEATURE_LEVEL_RECORD: - featureControl.replay((FeatureLevelRecord) message); - handleFeatureControlChange(); - break; - case CLIENT_QUOTA_RECORD: - clientQuotaControlManager.replay((ClientQuotaRecord) message); - break; - case PRODUCER_IDS_RECORD: - producerIdControlManager.replay((ProducerIdsRecord) message); - break; - case BROKER_REGISTRATION_CHANGE_RECORD: - clusterControl.replay((BrokerRegistrationChangeRecord) message); - break; - case ACCESS_CONTROL_ENTRY_RECORD: - aclControlManager.replay((AccessControlEntryRecord) message, snapshotId); - break; - case REMOVE_ACCESS_CONTROL_ENTRY_RECORD: - aclControlManager.replay((RemoveAccessControlEntryRecord) message, snapshotId); - break; - case NO_OP_RECORD: - // NoOpRecord is an empty record and doesn't need to be replayed - break; - default: - throw new RuntimeException("Unhandled record type " + type); - } - } catch (Exception e) { - if (snapshotId.isPresent()) { - log.error("Error replaying record {} from snapshot {} at last offset {}.", - message.toString(), snapshotId.get(), offset, e); - } else { - log.error("Error replaying record {} at last offset {}.", - message.toString(), offset, e); - } + private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId) { + MetadataRecordType type = MetadataRecordType.fromId(message.apiKey()); + switch (type) { + case REGISTER_BROKER_RECORD: + clusterControl.replay((RegisterBrokerRecord) message); + break; + case UNREGISTER_BROKER_RECORD: + clusterControl.replay((UnregisterBrokerRecord) message); + break; + case TOPIC_RECORD: + replicationControl.replay((TopicRecord) message); + break; + case PARTITION_RECORD: + replicationControl.replay((PartitionRecord) message); + break; + case CONFIG_RECORD: + configurationControl.replay((ConfigRecord) message); + break; + case PARTITION_CHANGE_RECORD: + replicationControl.replay((PartitionChangeRecord) message); + break; + case FENCE_BROKER_RECORD: + clusterControl.replay((FenceBrokerRecord) message); + break; + case UNFENCE_BROKER_RECORD: + clusterControl.replay((UnfenceBrokerRecord) message); + break; + case REMOVE_TOPIC_RECORD: + replicationControl.replay((RemoveTopicRecord) message); + break; + case FEATURE_LEVEL_RECORD: + featureControl.replay((FeatureLevelRecord) message); + handleFeatureControlChange(); + break; + case CLIENT_QUOTA_RECORD: + clientQuotaControlManager.replay((ClientQuotaRecord) message); + break; + case PRODUCER_IDS_RECORD: + producerIdControlManager.replay((ProducerIdsRecord) message); + break; + case BROKER_REGISTRATION_CHANGE_RECORD: + clusterControl.replay((BrokerRegistrationChangeRecord) message); + break; + case ACCESS_CONTROL_ENTRY_RECORD: + aclControlManager.replay((AccessControlEntryRecord) message, snapshotId); + break; + case REMOVE_ACCESS_CONTROL_ENTRY_RECORD: + aclControlManager.replay((RemoveAccessControlEntryRecord) message, snapshotId); + break; + case NO_OP_RECORD: + // NoOpRecord is an empty record and doesn't need to be replayed + break; + default: + throw new RuntimeException("Unhandled record type " + type); } } @@ -1344,8 +1394,24 @@ public final class QuorumController implements Controller { updateLastCommittedState(-1, -1, -1); } + /** + * Handles faults that should normally be fatal to the process. + */ + private final FaultHandler fatalFaultHandler; + + /** + * Handles faults in metadata handling that are normally not fatal. + */ + private final FaultHandler metadataFaultHandler; + + /** + * The slf4j log context, used to create new loggers. + */ private final LogContext logContext; + /** + * The slf4j logger. + */ private final Logger log; /** @@ -1530,28 +1596,34 @@ public final class QuorumController implements Controller { private final BootstrapMetadata bootstrapMetadata; - private QuorumController(LogContext logContext, - int nodeId, - String clusterId, - KafkaEventQueue queue, - Time time, - KafkaConfigSchema configSchema, - RaftClient<ApiMessageAndVersion> raftClient, - QuorumFeatures quorumFeatures, - short defaultReplicationFactor, - int defaultNumPartitions, - ReplicaPlacer replicaPlacer, - long snapshotMaxNewRecordBytes, - OptionalLong leaderImbalanceCheckIntervalNs, - OptionalLong maxIdleIntervalNs, - long sessionTimeoutNs, - ControllerMetrics controllerMetrics, - Optional<CreateTopicPolicy> createTopicPolicy, - Optional<AlterConfigPolicy> alterConfigPolicy, - ConfigurationValidator configurationValidator, - Optional<ClusterMetadataAuthorizer> authorizer, - Map<String, Object> staticConfig, - BootstrapMetadata bootstrapMetadata) { + private QuorumController( + FaultHandler fatalFaultHandler, + FaultHandler metadataFaultHandler, + LogContext logContext, + int nodeId, + String clusterId, + KafkaEventQueue queue, + Time time, + KafkaConfigSchema configSchema, + RaftClient<ApiMessageAndVersion> raftClient, + QuorumFeatures quorumFeatures, + short defaultReplicationFactor, + int defaultNumPartitions, + ReplicaPlacer replicaPlacer, + long snapshotMaxNewRecordBytes, + OptionalLong leaderImbalanceCheckIntervalNs, + OptionalLong maxIdleIntervalNs, + long sessionTimeoutNs, + ControllerMetrics controllerMetrics, + Optional<CreateTopicPolicy> createTopicPolicy, + Optional<AlterConfigPolicy> alterConfigPolicy, + ConfigurationValidator configurationValidator, + Optional<ClusterMetadataAuthorizer> authorizer, + Map<String, Object> staticConfig, + BootstrapMetadata bootstrapMetadata + ) { + this.fatalFaultHandler = fatalFaultHandler; + this.metadataFaultHandler = metadataFaultHandler; this.logContext = logContext; this.log = logContext.logger(QuorumController.class); this.nodeId = nodeId; diff --git a/metadata/src/main/java/org/apache/kafka/metadata/fault/MetadataFaultException.java b/metadata/src/main/java/org/apache/kafka/metadata/fault/MetadataFaultException.java new file mode 100644 index 0000000000..c57ce46fb3 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metadata/fault/MetadataFaultException.java @@ -0,0 +1,32 @@ +/* + * 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.metadata.fault; + + +/** + * A fault that we encountered while we replayed cluster metadata. + */ +public class MetadataFaultException extends RuntimeException { + public MetadataFaultException(String message, Throwable cause) { + super(message, cause); + } + + public MetadataFaultException(String message) { + super(message); + } +} diff --git a/metadata/src/main/java/org/apache/kafka/metadata/fault/MetadataFaultHandler.java b/metadata/src/main/java/org/apache/kafka/metadata/fault/MetadataFaultHandler.java new file mode 100644 index 0000000000..e9f71b80e6 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metadata/fault/MetadataFaultHandler.java @@ -0,0 +1,36 @@ +/* + * 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.metadata.fault; + +import org.apache.kafka.server.fault.FaultHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Handles faults in Kafka metadata management. + */ +public class MetadataFaultHandler implements FaultHandler { + private static final Logger log = LoggerFactory.getLogger(MetadataFaultHandler.class); + + @Override + public void handleFault(String failureMessage, Throwable cause) { + FaultHandler.logFailureMessage(log, failureMessage, cause); + throw new MetadataFaultException("Encountered metadata fault: " + failureMessage, cause); + } +} diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java index a62b1f682f..5e395cebcb 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java @@ -45,6 +45,7 @@ import org.apache.kafka.common.Uuid; import org.apache.kafka.common.errors.BrokerIdNotRegisteredException; import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; import org.apache.kafka.common.message.RequestHeaderData; +import org.apache.kafka.common.metadata.ConfigRecord; import org.apache.kafka.common.security.auth.KafkaPrincipal; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.common.config.ConfigResource; @@ -1181,6 +1182,30 @@ public class QuorumControllerTest { } } + @Test + public void testFatalMetadataReplayErrorOnActive() throws Throwable { + try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv(3, Optional.empty())) { + try (QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv(logEnv, b -> { + })) { + QuorumController active = controlEnv.activeController(); + CompletableFuture<Void> future = active.appendWriteEvent("errorEvent", + OptionalLong.empty(), () -> { + return ControllerResult.of(Collections.singletonList(new ApiMessageAndVersion( + new ConfigRecord(). + setName(null). + setResourceName(null). + setResourceType((byte) 255). + setValue(null), (short) 0)), null); + }); + assertThrows(ExecutionException.class, () -> future.get()); + assertEquals(NullPointerException.class, + controlEnv.fatalFaultHandler().firstException().getCause().getClass()); + controlEnv.fatalFaultHandler().setIgnore(true); + controlEnv.metadataFaultHandler().setIgnore(true); + } + } + } + private static void assertInitialLoadFuturesNotComplete(List<StandardAuthorizer> authorizers) { for (int i = 0; i < authorizers.size(); i++) { assertFalse(authorizers.get(i).initialLoadFuture().isDone(), diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java index 4cc45a9774..40dd21c88d 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java +++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java @@ -34,6 +34,7 @@ import org.apache.kafka.controller.QuorumController.Builder; import org.apache.kafka.metalog.LocalLogManagerTestEnv; import org.apache.kafka.raft.LeaderAndEpoch; import org.apache.kafka.server.common.MetadataVersion; +import org.apache.kafka.server.fault.MockFaultHandler; import org.apache.kafka.test.TestUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -44,6 +45,8 @@ public class QuorumControllerTestEnv implements AutoCloseable { private final List<QuorumController> controllers; private final LocalLogManagerTestEnv logEnv; + private final MockFaultHandler fatalFaultHandler = new MockFaultHandler("fatalFaultHandler"); + private final MockFaultHandler metadataFaultHandler = new MockFaultHandler("metadataFaultHandler"); public QuorumControllerTestEnv( LocalLogManagerTestEnv logEnv, @@ -84,6 +87,8 @@ public class QuorumControllerTestEnv implements AutoCloseable { sessionTimeoutMillis.ifPresent(timeout -> { builder.setSessionTimeoutNs(NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS)); }); + builder.setFatalFaultHandler(fatalFaultHandler); + builder.setMetadataFaultHandler(metadataFaultHandler); builderConsumer.accept(builder); this.controllers.add(builder.build()); } @@ -117,6 +122,14 @@ public class QuorumControllerTestEnv implements AutoCloseable { return controllers; } + public MockFaultHandler fatalFaultHandler() { + return fatalFaultHandler; + } + + public MockFaultHandler metadataFaultHandler() { + return metadataFaultHandler; + } + @Override public void close() throws InterruptedException { for (QuorumController controller : controllers) { @@ -125,5 +138,7 @@ public class QuorumControllerTestEnv implements AutoCloseable { for (QuorumController controller : controllers) { controller.close(); } + fatalFaultHandler.maybeRethrowFirstException(); + metadataFaultHandler.maybeRethrowFirstException(); } } diff --git a/server-common/src/main/java/org/apache/kafka/server/fault/FaultHandler.java b/server-common/src/main/java/org/apache/kafka/server/fault/FaultHandler.java new file mode 100644 index 0000000000..4c03eacc32 --- /dev/null +++ b/server-common/src/main/java/org/apache/kafka/server/fault/FaultHandler.java @@ -0,0 +1,58 @@ +/* + * 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.server.fault; + +import org.slf4j.Logger; + + +/** + * Handle a server fault. + */ +public interface FaultHandler { + /** + * Handle a fault. + * + * @param failureMessage The failure message to log. + */ + default void handleFault(String failureMessage) { + handleFault(failureMessage, null); + } + + /** + * Handle a fault. + * + * @param failureMessage The failure message to log. + * @param cause The exception that caused the problem, or null. + */ + void handleFault(String failureMessage, Throwable cause); + + /** + * Log a failure message about a fault. + * + * @param log The log4j logger. + * @param failureMessage The failure message. + * @param cause The exception which caused the failure, or null. + */ + static void logFailureMessage(Logger log, String failureMessage, Throwable cause) { + if (cause == null) { + log.error("Encountered fatal fault: {}", failureMessage); + } else { + log.error("Encountered fatal fault: {}", failureMessage, cause); + } + } +} diff --git a/server-common/src/main/java/org/apache/kafka/server/fault/ProcessExitingFaultHandler.java b/server-common/src/main/java/org/apache/kafka/server/fault/ProcessExitingFaultHandler.java new file mode 100644 index 0000000000..e3b9f25a3b --- /dev/null +++ b/server-common/src/main/java/org/apache/kafka/server/fault/ProcessExitingFaultHandler.java @@ -0,0 +1,37 @@ +/* + * 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.server.fault; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.kafka.common.utils.Exit; + + +/** + * This is a fault handler which exits the Java process. + */ +public class ProcessExitingFaultHandler implements FaultHandler { + private static final Logger log = LoggerFactory.getLogger(ProcessExitingFaultHandler.class); + + @Override + public void handleFault(String failureMessage, Throwable cause) { + FaultHandler.logFailureMessage(log, failureMessage, cause); + Exit.exit(1); + } +} diff --git a/server-common/src/test/java/org/apache/kafka/server/fault/MockFaultHandler.java b/server-common/src/test/java/org/apache/kafka/server/fault/MockFaultHandler.java new file mode 100644 index 0000000000..39b3ed0784 --- /dev/null +++ b/server-common/src/test/java/org/apache/kafka/server/fault/MockFaultHandler.java @@ -0,0 +1,65 @@ +/* + * 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.server.fault; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * This is a fault handler suitable for use in JUnit tests. It will store the result of the first + * call to handleFault that was made. + */ +public class MockFaultHandler implements FaultHandler { + private static final Logger log = LoggerFactory.getLogger(MockFaultHandler.class); + + private final String name; + private MockFaultHandlerException firstException = null; + private boolean ignore = false; + + public MockFaultHandler(String name) { + this.name = name; + } + + @Override + public synchronized void handleFault(String failureMessage, Throwable cause) { + FaultHandler.logFailureMessage(log, failureMessage, cause); + MockFaultHandlerException e = (cause == null) ? + new MockFaultHandlerException(name + ": " + failureMessage) : + new MockFaultHandlerException(name + ": " + failureMessage + + ": " + cause.getMessage(), cause); + if (firstException == null) { + firstException = e; + } + throw e; + } + + public synchronized void maybeRethrowFirstException() { + if (firstException != null && !ignore) { + throw firstException; + } + } + + public synchronized MockFaultHandlerException firstException() { + return firstException; + } + + public synchronized void setIgnore(boolean ignore) { + this.ignore = ignore; + } +} diff --git a/server-common/src/test/java/org/apache/kafka/server/fault/MockFaultHandlerException.java b/server-common/src/test/java/org/apache/kafka/server/fault/MockFaultHandlerException.java new file mode 100644 index 0000000000..ef9b11bdeb --- /dev/null +++ b/server-common/src/test/java/org/apache/kafka/server/fault/MockFaultHandlerException.java @@ -0,0 +1,38 @@ +/* + * 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.server.fault; + + +/** + * An exception thrown by MockFaultHandler. + */ +public class MockFaultHandlerException extends RuntimeException { + public MockFaultHandlerException(String failureMessage, Throwable cause) { + super(failureMessage, cause); + // If a cause exception was provided, set our the stack trace its stack trace. This is + // useful in junit tests where a limited number of stack frames are printed, and usually + // the stack frames of cause exceptions get trimmed. + if (cause != null) { + setStackTrace(cause.getStackTrace()); + } + } + + public MockFaultHandlerException(String failureMessage) { + this(failureMessage, null); + } +}