test(qa): add reproducible test to verify multiple rejections are han…

…dled

(cherry picked from commit 3780629)

qa/integration-tests/src/test/java/io/camunda/zeebe/it/processing/MultipleInvalidResourceDeletionTest.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.it.processing;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.camunda.zeebe.client.ZeebeClient;
+import io.camunda.zeebe.client.api.command.ClientStatusException;
+import io.camunda.zeebe.protocol.Protocol;
+import io.camunda.zeebe.qa.util.actuator.PartitionsActuator;
+import io.camunda.zeebe.qa.util.cluster.TestStandaloneBroker;
+import io.camunda.zeebe.qa.util.junit.ZeebeIntegration;
+import io.camunda.zeebe.qa.util.junit.ZeebeIntegration.TestZeebe;
+import io.camunda.zeebe.test.util.junit.AutoCloseResources.AutoCloseResource;
+import io.camunda.zeebe.test.util.junit.RegressionTest;
+import io.grpc.Status.Code;
+import org.junit.jupiter.api.BeforeEach;
+
+@ZeebeIntegration
+public class MultipleInvalidResourceDeletionTest {
+
+  @TestZeebe
+  private final TestStandaloneBroker zeebe =
+      new TestStandaloneBroker()
+          // disable long polling to increase the load in streamprocessor
+          .withGatewayConfig(cfg -> cfg.getLongPolling().setEnabled(false))
+          .withRecordingExporter(true);
+
+  private final PartitionsActuator partitions = PartitionsActuator.of(zeebe);
+  @AutoCloseResource private ZeebeClient client;
+
+  @BeforeEach
+  void beforeEach() {
+    client = zeebe.newClientBuilder().build();
+  }
+
+  @RegressionTest("https://github.com/camunda/zeebe/issues/16429")
+  public void shouldRejectMultipleResourceDeletion() {
+    // given
+    // generate load on stream processor. This is to increase the chance of the issue to happen.
+    client
+        .newWorker()
+        .jobType("test")
+        .handler(
+            (jobClient, job) -> {
+              jobClient.newCompleteCommand(job.getKey()).send().join();
+            })
+        .open();
+
+    // when
+    final var firstRejection =
+        client.newDeleteResourceCommand(Protocol.encodePartitionId(1, 999999L)).send();
+    final var secondRejection =
+        client.newDeleteResourceCommand(Protocol.encodePartitionId(1, 999999L)).send();
+    final var thirdRejection =
+        client.newDeleteResourceCommand(Protocol.encodePartitionId(1, 999999L)).send();
+
+    // then
+    // All requests should be rejected. In the original issue, the second/third requests can timeout
+    assertThatThrownBy(firstRejection::join)
+        .isInstanceOf(ClientStatusException.class)
+        .extracting("status.code")
+        .isEqualTo(Code.NOT_FOUND);
+
+    assertThatThrownBy(secondRejection::join)
+        .isInstanceOf(ClientStatusException.class)
+        .extracting("status.code")
+        .isEqualTo(Code.NOT_FOUND);
+
+    assertThatThrownBy(thirdRejection::join)
+        .isInstanceOf(ClientStatusException.class)
+        .extracting("status.code")
+        .isEqualTo(Code.NOT_FOUND);
+  }
+}