fix: stop activating jobs if request isn't open anymore

(cherry picked from commit cdec74d)

gateway/src/main/java/io/camunda/zeebe/gateway/impl/job/InflightActivateJobsRequest.java

@@ -32,6 +32,7 @@ public final class InflightActivateJobsRequest {
   private ScheduledTimer scheduledTimer;
   private boolean isTimedOut;
   private boolean isCompleted;
+  private boolean isAborted;
 
   public InflightActivateJobsRequest(
       final long requestId,
@@ -66,7 +67,7 @@ private InflightActivateJobsRequest(
   }
 
   public void complete() {
-    if (isCompleted() || isCanceled()) {
+    if (!isOpen()) {
       return;
     }
     cancelTimerIfScheduled();
@@ -83,7 +84,7 @@ public boolean isCompleted() {
   }
 
   public void onResponse(final ActivateJobsResponse grpcResponse) {
-    if (!(isCompleted() || isCanceled())) {
+    if (isOpen()) {
       try {
         responseObserver.onNext(grpcResponse);
       } catch (final Exception e) {
@@ -93,15 +94,16 @@ public void onResponse(final ActivateJobsResponse grpcResponse) {
   }
 
   public void onError(final Throwable error) {
-    if (isCompleted() || isCanceled()) {
+    if (!isOpen()) {
       return;
     }
     cancelTimerIfScheduled();
     try {
       responseObserver.onError(error);
     } catch (final Exception e) {
-      LOG.warn("Failed to send response to client.", e);
+      LOG.warn("Failed to send terminating error to client.", e);
     }
+    isAborted = true;
   }
 
   public void timeout() {
@@ -163,6 +165,14 @@ private void cancelTimerIfScheduled() {
     }
   }
 
+  public boolean isAborted() {
+    return isAborted;
+  }
+
+  public boolean isOpen() {
+    return !(isCompleted() || isCanceled() || isAborted());
+  }
+
   @Override
   public int hashCode() {
     return Objects.hash(jobType, maxJobsToActivate, requestId, worker);

gateway/src/main/java/io/camunda/zeebe/gateway/impl/job/RoundRobinActivateJobsHandler.java

@@ -93,6 +93,10 @@ private void activateJobs(
       final ResponseObserverDelegate delegate) {
     actor.run(
         () -> {
+          if (!request.isOpen()) {
+            return;
+          }
+
           if (requestState.shouldActivateJobs()) {
             final var brokerRequest = request.getRequest();
             final var partitionId = requestState.getNextPartition();