KAFKA-20723: Add fail fast check for share group dlq(...). [1/N] (apache#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 <apoorvmittal10@gmail.com>, Andrew Schofield
 <aschofield@confluent.io>

Author: smjn

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())
 

server/src/main/java/org/apache/kafka/server/share/dlq/DefaultShareGroupDLQManager.java

@@ -62,22 +62,22 @@ private DefaultShareGroupDLQManager(
         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,16 +111,16 @@ private static void validate(ShareGroupDLQRecordParameter param) {
             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.");
         }
 
         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.");
+        }
     }
 }

server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java

@@ -185,6 +185,9 @@ public CompletableFuture<Void> dlq(ShareGroupDLQRecordParameter param) {
 
     // 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);

server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java

@@ -382,7 +382,33 @@ public void testStopWithoutStartIsNoOp() {
         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 @@ public void testDlqEmptyTopicNameFailsValidation() throws Exception {
         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 @@ public void testDlqTopicStartingWithUnderscoreFailsValidation() throws Exception
         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 @@ public void testDlqExistingTopicWithoutDlqConfigFailsValidation() throws Excepti
         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 @@ public void testDlqTopicMissingAndAutoCreateDisabledFailsValidation() throws Exc
         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,24 +479,28 @@ public void testDlqTopicPrefixMismatchFailsValidation() throws Exception {
         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"));
         verifyNoInteractions(mockMetrics);
     }
 
     @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);
     }