Add SAC test with different groups on same super stream

References #46, rabbitmq/rabbitmq-server#3753

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/StreamConsumer.java

@@ -123,12 +123,25 @@ class StreamConsumer implements Consumer {
       if (Utils.isSac(subscriptionProperties)) {
         this.sacStatus = ConsumerUpdateListener.Status.PASSIVE;
         MessageHandler existingMessageHandler = decoratedMessageHandler.get();
-        MessageHandler messageHandlerWithSac =
-            (context, message) -> {
-              if (sacStatus == ConsumerUpdateListener.Status.ACTIVE) {
-                existingMessageHandler.handle(context, message);
-              }
-            };
+        AtomicBoolean receivedSomething = new AtomicBoolean(false);
+        MessageHandler messageHandlerWithSac;
+        if (trackingConfiguration.auto()) {
+          messageHandlerWithSac =
+              (context, message) -> {
+                if (sacStatus == ConsumerUpdateListener.Status.ACTIVE) {
+                  receivedSomething.set(true);
+                  existingMessageHandler.handle(context, message);
+                }
+              };
+        } else {
+          messageHandlerWithSac =
+              (context, message) -> {
+                if (sacStatus == ConsumerUpdateListener.Status.ACTIVE) {
+                  existingMessageHandler.handle(context, message);
+                }
+              };
+        }
+
         decoratedMessageHandler.set(messageHandlerWithSac);
 
         if (consumerUpdateListener == null
@@ -160,17 +173,21 @@ class StreamConsumer implements Consumer {
                     return result;
                   } else if (context.previousStatus() == ConsumerUpdateListener.Status.ACTIVE
                       && context.status() == ConsumerUpdateListener.Status.PASSIVE) {
-                    LOGGER.debug(
-                        "Storing offset (consumer {}, stream {}) because going from active to passive",
-                        this.id,
-                        this.stream);
-                    long offset = trackingFlushCallback.getAsLong();
-                    LOGGER.debug(
-                        "Making sure offset {} has been stored (consumer {}, stream {})",
-                        offset,
-                        this.id,
-                        this.stream);
-                    waitForOffsetToBeStored(offset);
+                    if (receivedSomething.get()) {
+                      LOGGER.debug(
+                          "Storing offset (consumer {}, stream {}) because going from active to passive",
+                          this.id,
+                          this.stream);
+                      long offset = trackingFlushCallback.getAsLong();
+                      LOGGER.debug(
+                          "Making sure offset {} has been stored (consumer {}, stream {})",
+                          offset,
+                          this.id,
+                          this.stream);
+                      waitForOffsetToBeStored(offset);
+                    } else {
+                      LOGGER.debug("Nothing received yet, no need to store offset");
+                    }
                     result = OffsetSpecification.none();
                   }
                   return result;
@@ -292,7 +309,7 @@ void waitForOffsetToBeStored(long expectedStoredOffset) {
                 this.name, this.stream, expectedStoredOffset)
             .delayPolicy(
                 BackOffDelayPolicy.fixedWithInitialDelay(
-                    Duration.ofMillis(500), Duration.ofMillis(200)))
+                    Duration.ofMillis(200), Duration.ofMillis(200)))
             .retry(exception -> exception instanceof IllegalStateException)
             .scheduler(environment.scheduledExecutorService())
             .build();

src/test/java/com/rabbitmq/stream/impl/SacClientTest.java

@@ -318,6 +318,7 @@ void singleActiveConsumersShouldSpreadOnSuperStreamPartitions(TestInfo info) thr
     String superStream = streamName(info);
     String consumerName = "foo";
     Connection c = new ConnectionFactory().newConnection();
+    // subscription distribution
     // client 1: 0, 1, 2 / client 2: 3, 4, 5, / client 3: 6, 7, 8
     try {
       declareSuperStreamTopology(c, superStream, 3);

src/test/java/com/rabbitmq/stream/impl/SacSuperStreamConsumerTest.java

@@ -35,6 +35,8 @@
 import com.rabbitmq.stream.impl.Utils.CompositeConsumerUpdateListener;
 import io.netty.channel.EventLoopGroup;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -728,6 +730,87 @@ void sacCustomOffsetTrackingShouldTakeOverOnRelanbancing() throws Exception {
                 .isGreaterThanOrEqualTo(expectedMessageCountPerPartition - 1));
   }
 
+  @Test
+  void consumerGroupsOnSameSuperStreamShouldBeIndependent() throws Exception {
+    declareSuperStreamTopology(connection, superStream, partitionCount);
+    int messageCount = 20_000;
+    int consumerGroupsCount = 3;
+    List<String> consumerNames =
+        IntStream.range(0, consumerGroupsCount).mapToObj(i -> "my-app-" + i).collect(toList());
+    List<String> partitions =
+        IntStream.range(0, partitionCount).mapToObj(i -> superStream + "-" + i).collect(toList());
+    Map<String, AtomicInteger> receivedMessages = new ConcurrentHashMap<>(consumerGroupsCount);
+    Map<String, AtomicInteger> receivedMessagesPerPartitions =
+        new ConcurrentHashMap<>(consumerGroupsCount * partitionCount);
+    Map<String, Long> lastReceivedOffsets = new ConcurrentHashMap<>();
+    partitions.forEach(
+        partition -> {
+          lastReceivedOffsets.put(partition, 0L);
+          receivedMessagesPerPartitions.put(partition, new AtomicInteger(0));
+        });
+
+    List<Consumer> consumers = new ArrayList<>(consumerGroupsCount * partitionCount);
+    consumerNames.forEach(
+        consumerName -> {
+          receivedMessages.put(consumerName, new AtomicInteger(0));
+          partitions.forEach(
+              partition -> {
+                receivedMessagesPerPartitions.put(partition + consumerName, new AtomicInteger(0));
+                Consumer consumer =
+                    environment
+                        .consumerBuilder()
+                        .singleActiveConsumer()
+                        .superStream(superStream)
+                        .offset(OffsetSpecification.first())
+                        .name(consumerName)
+                        .autoTrackingStrategy()
+                        .builder()
+                        .messageHandler(
+                            (context, message) -> {
+                              lastReceivedOffsets.put(
+                                  context.stream() + consumerName, context.offset());
+                              receivedMessagesPerPartitions
+                                  .get(context.stream() + consumerName)
+                                  .incrementAndGet();
+                              receivedMessages.get(consumerName).incrementAndGet();
+                            })
+                        .build();
+                consumers.add(consumer);
+              });
+        });
+
+    partitions.forEach(
+        partition -> TestUtils.publishAndWaitForConfirms(cf, messageCount, partition));
+
+    int messageTotal = messageCount * partitionCount;
+    consumerNames.forEach(
+        consumerName -> waitUntil(() -> receivedMessages.get(consumerName).get() == messageTotal));
+
+    consumerNames.forEach(
+        consumerName -> {
+          // summing the received messages for the consumer group name
+          assertThat(
+                  receivedMessagesPerPartitions.entrySet().stream()
+                      .filter(e -> e.getKey().endsWith(consumerName))
+                      .mapToInt(e -> e.getValue().get())
+                      .sum())
+              .isEqualTo(messageTotal);
+        });
+
+    Collections.reverse(consumers);
+    consumers.forEach(Consumer::close);
+
+    Client c = cf.get();
+    consumerNames.forEach(
+        consumerName -> {
+          partitions.forEach(
+              partition -> {
+                assertThat(c.queryOffset(consumerName, partition).getOffset())
+                    .isEqualTo(lastReceivedOffsets.get(partition + consumerName));
+              });
+        });
+  }
+
   private static void waitUntil(CallableBooleanSupplier action) {
     try {
       waitAtMost(action);