This is an automated email from the ASF dual-hosted git repository.
apoorvmittal10 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 28ec585cb3c KAFKA-20723: Add fail fast check for share group dlq(...).
[1/N] (#22635)
28ec585cb3c is described below
commit 28ec585cb3cf4a6510a8c118464da4934049273c
Author: Sushant Mahajan <[email protected]>
AuthorDate: Wed Jun 24 18:40:25 2026 +0530
KAFKA-20723: Add fail fast check for share group dlq(...). [1/N] (#22635)
* Add check whether the DLQ state manager is started, before enqueuing
the request in the `Sender`.
* Change sequence of validations in `DefaultShareGroupDLQManager`.
* Add tests to verify the `dlq()` method changes.
Reviewers: Apoorv Mittal <[email protected]>, Andrew Schofield
<[email protected]>
---
.../src/main/scala/kafka/server/BrokerServer.scala | 7 ++++
.../share/dlq/DefaultShareGroupDLQManager.java | 16 ++++-----
.../share/dlq/ShareGroupDLQStateManager.java | 3 ++
.../share/dlq/ShareGroupDLQStateManagerTest.java | 40 ++++++++++++++++++++--
4 files changed, 55 insertions(+), 11 deletions(-)
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala
b/core/src/main/scala/kafka/server/BrokerServer.scala
index 7af4c50da83..d7502aaa039 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -929,9 +929,16 @@ class BrokerServer(
Utils.closeQuietly(brokerTopicStats, "broker topic stats")
Utils.closeQuietly(sharePartitionManager, "share partition manager")
+ // The order of closing sharePartitionManager, groupCoordinator and
persister matters.
+ // groupCoordinator, sharePartitionManager must be closed before the
persister so that
+ // new requests from sharePartitionManager, groupCoordinator do not
encounter a stopped
+ // persister.
if (persister != null)
Utils.swallow(this.logger.underlying, () => persister.stop())
+ // The order of closing sharePartitionManager and shareGroupDLQManager
matters.
+ // sharePartitionManager must be closed before the shareGroupDLQManager
so any new
+ // requests from sharePartitionManager do not encounter a stopped
shareGroupDLQManager.
if (shareGroupDLQManager != null)
Utils.swallow(this.logger.underlying, () =>
shareGroupDLQManager.stop())
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
index 73559b3cebf..1c78a33ae72 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java
@@ -62,22 +62,22 @@ public class DefaultShareGroupDLQManager implements
ShareGroupDLQManager {
ShareGroupMetrics shareGroupMetrics,
LogReader logReader
) {
- this.stateManager = new ShareGroupDLQStateManager(client, cacheHelper,
time, timer, shareGroupMetrics, logReader);
+ stateManager = new ShareGroupDLQStateManager(client, cacheHelper,
time, timer, shareGroupMetrics, logReader);
}
private void start() {
- this.stateManager.start();
+ stateManager.start();
}
@Override
public CompletableFuture<Void> enqueue(ShareGroupDLQRecordParameter param)
{
try {
validate(param);
+ return stateManager.dlq(param);
} catch (Exception e) {
- log.error("Unable to validate dlq record parameters", e);
+ log.error("Unable to enqueue DLQ request", e);
return CompletableFuture.failedFuture(e);
}
- return stateManager.dlq(param);
}
@Override
@@ -111,10 +111,6 @@ public class DefaultShareGroupDLQManager implements
ShareGroupDLQManager {
throw new IllegalArgumentException(prefix + " partition cannot be
negative.");
}
- if (param.lastOffset() < param.firstOffset()) {
- throw new IllegalArgumentException(prefix + " last offset cannot
be less than first offset.");
- }
-
if (param.firstOffset() < 0) {
throw new IllegalArgumentException(prefix + " first offset cannot
be negative.");
}
@@ -122,5 +118,9 @@ public class DefaultShareGroupDLQManager implements
ShareGroupDLQManager {
if (param.lastOffset() < 0) {
throw new IllegalArgumentException(prefix + " last offset cannot
be negative.");
}
+
+ if (param.lastOffset() < param.firstOffset()) {
+ throw new IllegalArgumentException(prefix + " last offset cannot
be less than first offset.");
+ }
}
}
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
index 6bd1863aca8..8a0b3e04d26 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
@@ -185,6 +185,9 @@ public class ShareGroupDLQStateManager {
// Visibility for tests
CompletableFuture<Void> dlq(ShareGroupDLQRecordParameter param, long
requestBackoffMs, long requestBackoffMaxMs, int maxRequestAttempts) {
+ if (!this.isStarted.get()) {
+ return CompletableFuture.failedFuture(new
IllegalStateException("ShareGroupDLQStateManager is not started."));
+ }
CompletableFuture<Void> future = new CompletableFuture<>();
ProduceRequestHandler requestHandler = new
ProduceRequestHandler(param, future, requestBackoffMs, requestBackoffMaxMs,
maxRequestAttempts);
enqueue(requestHandler);
diff --git
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
index 2da51e250d3..a1767ab634b 100644
---
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
+++
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
@@ -382,7 +382,33 @@ class ShareGroupDLQStateManagerTest {
verifyNoInteractions(mockMetrics);
}
- // ---- DLQ topic validation tests (no thread start required) ----
+ @Test
+ public void testDlqBeforeStartFailsWithIllegalState() {
+ stateManager = builder().build();
+ // dlq() is invoked without a prior start(); the lifecycle guard must
reject it with a
+ // failed future rather than enqueueing onto a sender thread that is
not running.
+ CompletableFuture<Void> result = stateManager.dlq(param());
+ assertTrue(result.isDone());
+ assertTrue(result.isCompletedExceptionally());
+ assertInstanceOf(IllegalStateException.class, getCause(result));
+ verifyNoInteractions(mockMetrics);
+ }
+
+ @Test
+ public void testDlqAfterStopFailsWithIllegalState() throws Exception {
+ stateManager = builder().build();
+ stateManager.start();
+ stateManager.stop();
+ // Once stopped, dlq() must fail fast rather than enqueueing onto a
shut-down sender thread
+ // where the future would never complete.
+ CompletableFuture<Void> result = stateManager.dlq(param());
+ assertTrue(result.isDone());
+ assertTrue(result.isCompletedExceptionally());
+ assertInstanceOf(IllegalStateException.class, getCause(result));
+ verifyNoInteractions(mockMetrics);
+ }
+
+ // ---- DLQ topic validation tests ----
@Test
public void testDlqEmptyTopicNameFailsValidation() throws Exception {
@@ -391,6 +417,7 @@ class ShareGroupDLQStateManagerTest {
when(cacheHelper.shareGroupDlqTopicPrefix()).thenReturn(Optional.empty());
stateManager = builder().withCacheHelper(cacheHelper).build();
+ stateManager.start();
Throwable cause = getCause(stateManager.dlq(param()));
assertInstanceOf(ConfigException.class, cause);
assertTrue(cause.getMessage().contains("empty"));
@@ -404,6 +431,7 @@ class ShareGroupDLQStateManagerTest {
when(cacheHelper.shareGroupDlqTopicPrefix()).thenReturn(Optional.empty());
stateManager = builder().withCacheHelper(cacheHelper).build();
+ stateManager.start();
Throwable cause = getCause(stateManager.dlq(param()));
assertInstanceOf(ConfigException.class, cause);
assertTrue(cause.getMessage().contains("__"));
@@ -419,6 +447,7 @@ class ShareGroupDLQStateManagerTest {
when(cacheHelper.isDlqEnabledOnTopic(DLQ_TOPIC)).thenReturn(false);
stateManager = builder().withCacheHelper(cacheHelper).build();
+ stateManager.start();
Throwable cause = getCause(stateManager.dlq(param()));
assertInstanceOf(ConfigException.class, cause);
assertTrue(cause.getMessage().contains("DLQ is not enabled"));
@@ -434,6 +463,7 @@ class ShareGroupDLQStateManagerTest {
when(cacheHelper.isDlqAutoTopicCreateEnabled()).thenReturn(false);
stateManager = builder().withCacheHelper(cacheHelper).build();
+ stateManager.start();
Throwable cause = getCause(stateManager.dlq(param()));
assertInstanceOf(ConfigException.class, cause);
assertTrue(cause.getMessage().contains("auto create is disabled"));
@@ -449,6 +479,7 @@ class ShareGroupDLQStateManagerTest {
when(cacheHelper.isDlqEnabledOnTopic(DLQ_TOPIC)).thenReturn(true);
stateManager = builder().withCacheHelper(cacheHelper).build();
+ stateManager.start();
Throwable cause = getCause(stateManager.dlq(param()));
assertInstanceOf(ConfigException.class, cause);
assertTrue(cause.getMessage().contains("does not comply with the DLQ
topic prefix"));
@@ -456,17 +487,20 @@ class ShareGroupDLQStateManagerTest {
}
@Test
- public void testDlqValidationFailureCompletesFutureBeforeStart() throws
Exception {
+ public void testDlqValidationFailureCompletesFutureSynchronously() throws
Exception {
ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
when(cacheHelper.shareGroupDlqTopic(GROUP_ID)).thenReturn(Optional.empty());
when(cacheHelper.shareGroupDlqTopicPrefix()).thenReturn(Optional.empty());
- // validateDlqTopic runs synchronously inside dlq(), so it should fail
without the sender thread.
+ // validateDlqTopic runs synchronously inside dlq() on the calling
thread, so a validation
+ // failure completes the returned future before dlq() returns - no
sender-thread round trip.
stateManager = builder().withCacheHelper(cacheHelper).build();
+ stateManager.start();
CompletableFuture<Void> result = stateManager.dlq(param());
assertTrue(result.isDone());
assertTrue(result.isCompletedExceptionally());
assertFalse(result.isCancelled());
+ assertInstanceOf(ConfigException.class, getCause(result));
verifyNoInteractions(mockMetrics);
}
