merge: #9152

9152: [Backport stable/8.0] fix: remove subscription consumer instead of re-registering r=oleschoenburg a=github-actions[bot]

# Description
Backport of #9139 to `stable/8.0`.

relates to #9123

Co-authored-by: Ole Schönburg <ole.schoenburg@gmail.com>

dispatcher/src/main/java/io/camunda/zeebe/dispatcher/Subscription.java

@@ -82,7 +82,7 @@ public void registerConsumer(final ActorCondition consumer) {
 
   @Override
   public void removeConsumer(final ActorCondition consumer) {
-    actorConditions.registerConsumer(consumer);
+    actorConditions.removeConsumer(consumer);
   }
 
   protected long getLimit() {

dispatcher/src/test/java/io/camunda/zeebe/dispatcher/SubscriptionConsumerTest.java

@@ -0,0 +1,62 @@
+/*
+ * 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.dispatcher;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+import io.camunda.zeebe.dispatcher.impl.log.LogBuffer;
+import io.camunda.zeebe.util.sched.ActorCondition;
+import org.junit.jupiter.api.Test;
+
+final class SubscriptionConsumerTest {
+
+  @Test
+  void consumersAreSignaledAfterRegistering() {
+    // given
+    final var consumer = mock(ActorCondition.class);
+    final var subscription =
+        new Subscription(
+            mock(AtomicPosition.class),
+            mock(AtomicPosition.class),
+            0,
+            "",
+            mock(ActorCondition.class),
+            mock(LogBuffer.class));
+
+    // when
+    subscription.registerConsumer(consumer);
+
+    // then
+    subscription.getActorConditions().signalConsumers();
+    verify(consumer).signal();
+  }
+
+  @Test
+  void consumersAreNotSignaledAfterRemoving() {
+    // given
+    final var consumer = mock(ActorCondition.class);
+    final var subscription =
+        new Subscription(
+            mock(AtomicPosition.class),
+            mock(AtomicPosition.class),
+            0,
+            "",
+            mock(ActorCondition.class),
+            mock(LogBuffer.class));
+    subscription.registerConsumer(consumer);
+
+    // when
+    subscription.removeConsumer(consumer);
+
+    // then
+    subscription.getActorConditions().signalConsumers();
+    verifyNoInteractions(consumer);
+  }
+}