You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2023/05/15 08:26:20 UTC
[kafka] branch 3.4 updated: MINOR: fix some flaky KRaft-related tests (#13543) (#13543)
This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch 3.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.4 by this push:
new 8f89a12c8ee MINOR: fix some flaky KRaft-related tests (#13543) (#13543)
8f89a12c8ee is described below
commit 8f89a12c8ee8db7badab76ca16a2d27ab30fd29f
Author: Colin Patrick McCabe <cm...@apache.org>
AuthorDate: Fri Apr 14 10:39:08 2023 -0700
MINOR: fix some flaky KRaft-related tests (#13543) (#13543)
In SharedServer, fix some cases where a volatile variable could change to null while we were using
it, during shutdown. This is mainly a junit test issue, although it could also cause ugly error
messages during shutdown when running the server in a production context.
Fix a race in KafkaEventQueueTest.testSize.
Reviewers: David Arthur <mu...@gmail.com>
---
.../src/main/scala/kafka/server/SharedServer.scala | 31 +++++++++++-----------
.../apache/kafka/queue/KafkaEventQueueTest.java | 4 +--
2 files changed, 18 insertions(+), 17 deletions(-)
diff --git a/core/src/main/scala/kafka/server/SharedServer.scala b/core/src/main/scala/kafka/server/SharedServer.scala
index 97d4a0e470f..55c5105603e 100644
--- a/core/src/main/scala/kafka/server/SharedServer.scala
+++ b/core/src/main/scala/kafka/server/SharedServer.scala
@@ -165,8 +165,8 @@ class SharedServer(
name = "metadata loading",
fatal = sharedServerConfig.processRoles.contains(ControllerRole),
action = () => SharedServer.this.synchronized {
- if (brokerMetrics != null) brokerMetrics.metadataLoadErrorCount.getAndIncrement()
- if (controllerMetrics != null) controllerMetrics.incrementMetadataErrorCount()
+ Option(brokerMetrics).foreach(_.metadataLoadErrorCount.getAndIncrement())
+ Option(controllerMetrics).foreach(_.incrementMetadataErrorCount())
snapshotsDiabledReason.compareAndSet(null, "metadata loading fault")
})
@@ -177,8 +177,8 @@ class SharedServer(
name = "initial broker metadata loading",
fatal = true,
action = () => SharedServer.this.synchronized {
- if (brokerMetrics != null) brokerMetrics.metadataApplyErrorCount.getAndIncrement()
- if (controllerMetrics != null) controllerMetrics.incrementMetadataErrorCount()
+ Option(brokerMetrics).foreach(_.metadataApplyErrorCount.getAndIncrement())
+ Option(controllerMetrics).foreach(_.incrementMetadataErrorCount())
snapshotsDiabledReason.compareAndSet(null, "initial broker metadata loading fault")
})
@@ -189,7 +189,7 @@ class SharedServer(
name = "quorum controller",
fatal = true,
action = () => SharedServer.this.synchronized {
- if (controllerMetrics != null) controllerMetrics.incrementMetadataErrorCount()
+ Option(controllerMetrics).foreach(_.incrementMetadataErrorCount())
snapshotsDiabledReason.compareAndSet(null, "quorum controller fault")
})
@@ -200,8 +200,8 @@ class SharedServer(
name = "metadata publishing",
fatal = false,
action = () => SharedServer.this.synchronized {
- if (brokerMetrics != null) brokerMetrics.metadataApplyErrorCount.getAndIncrement()
- if (controllerMetrics != null) controllerMetrics.incrementMetadataErrorCount()
+ Option(brokerMetrics).foreach(_.metadataApplyErrorCount.getAndIncrement())
+ Option(controllerMetrics).foreach(_.incrementMetadataErrorCount())
// Note: snapshot generation does not need to be disabled for a publishing fault.
})
@@ -224,7 +224,7 @@ class SharedServer(
if (sharedServerConfig.processRoles.contains(ControllerRole)) {
controllerMetrics = new QuorumControllerMetrics(KafkaYammerMetrics.defaultRegistry(), time)
}
- raftManager = new KafkaRaftManager[ApiMessageAndVersion](
+ val _raftManager = new KafkaRaftManager[ApiMessageAndVersion](
metaProps,
sharedServerConfig,
new MetadataRecordSerde,
@@ -236,7 +236,8 @@ class SharedServer(
controllerQuorumVotersFuture,
raftManagerFaultHandler
)
- raftManager.startup()
+ raftManager = _raftManager
+ _raftManager.startup()
if (sharedServerConfig.processRoles.contains(ControllerRole)) {
val loaderBuilder = new MetadataLoader.Builder().
@@ -244,14 +245,14 @@ class SharedServer(
setTime(time).
setThreadNamePrefix(threadNamePrefix.getOrElse("")).
setFaultHandler(metadataLoaderFaultHandler).
- setHighWaterMarkAccessor(() => raftManager.client.highWatermark())
+ setHighWaterMarkAccessor(() => _raftManager.client.highWatermark())
if (brokerMetrics != null) {
loaderBuilder.setMetadataLoaderMetrics(brokerMetrics)
}
loader = loaderBuilder.build()
snapshotEmitter = new SnapshotEmitter.Builder().
setNodeId(metaProps.nodeId).
- setRaftClient(raftManager.client).
+ setRaftClient(_raftManager.client).
build()
snapshotGenerator = new SnapshotGenerator.Builder(snapshotEmitter).
setNodeId(metaProps.nodeId).
@@ -261,7 +262,7 @@ class SharedServer(
setMaxTimeSinceLastSnapshotNs(TimeUnit.MILLISECONDS.toNanos(sharedServerConfig.metadataSnapshotMaxIntervalMs)).
setDisabledReason(snapshotsDiabledReason).
build()
- raftManager.register(loader)
+ _raftManager.register(loader)
try {
loader.installPublishers(Collections.singletonList(snapshotGenerator))
} catch {
@@ -285,10 +286,10 @@ class SharedServer(
def ensureNotRaftLeader(): Unit = synchronized {
// Ideally, this would just resign our leadership, if we had it. But we don't have an API in
// RaftManager for that yet, so shut down the RaftManager.
- if (raftManager != null) {
- CoreUtils.swallow(raftManager.shutdown(), this)
+ Option(raftManager).foreach(_raftManager => {
+ CoreUtils.swallow(_raftManager.shutdown(), this)
raftManager = null
- }
+ })
}
private def stop(): Unit = synchronized {
diff --git a/server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java b/server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java
index 09910200a1f..7f04a188724 100644
--- a/server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java
+++ b/server-common/src/test/java/org/apache/kafka/queue/KafkaEventQueueTest.java
@@ -266,8 +266,8 @@ public class KafkaEventQueueTest {
assertFalse(queue.isEmpty());
queue.cancelDeferred("later");
queue.cancelDeferred("soon");
- assertTrue(queue.isEmpty());
+ TestUtils.waitForCondition(() -> queue.isEmpty(), "Failed to see the queue become empty.");
queue.close();
assertTrue(queue.isEmpty());
}
-}
\ No newline at end of file
+}