Add SAC test with custom consumer update listener

To mimick custom offset tracking.

References rabbitmq/rabbitmq-server#3753

Author: acogoluegnes

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

@@ -85,6 +85,64 @@ void autoTrackingSecondConsumerShouldTakeOverWhereTheFirstOneLeftOff() throws Ex
                   receivedMessages.get(1).incrementAndGet();
                 })
             .offset(OffsetSpecification.first())
+            .autoTrackingStrategy()
+            .builder()
+            .build();
+
+    publishAndWaitForConfirms(cf, messageCount, stream);
+    waitAtMost(() -> receivedMessages.getOrDefault(0, new AtomicInteger(0)).get() == messageCount);
+
+    assertThat(lastReceivedOffset).hasPositiveValue();
+    assertThat(receivedMessages.get(1)).hasValue(0);
+
+    long firstWaveLimit = lastReceivedOffset.get();
+    consumer1.close();
+
+    publishAndWaitForConfirms(cf, messageCount, stream);
+
+    waitAtMost(() -> receivedMessages.getOrDefault(0, new AtomicInteger(1)).get() == messageCount);
+    assertThat(lastReceivedOffset).hasValueGreaterThan(firstWaveLimit);
+    assertThat(receivedMessages.get(0)).hasValue(messageCount);
+
+    consumer2.close();
+  }
+
+  @Test
+  void customTrackingSecondConsumerShouldTakeOverWhereTheFirstOneLeftOff() throws Exception {
+    int messageCount = 10000;
+    Map<Integer, AtomicInteger> receivedMessages = new ConcurrentHashMap<>();
+    receivedMessages.put(0, new AtomicInteger(0));
+    receivedMessages.put(1, new AtomicInteger(0));
+    AtomicLong lastReceivedOffset = new AtomicLong(0);
+    String consumerName = "foo";
+    Consumer consumer1 =
+        environment.consumerBuilder().stream(stream)
+            .name(consumerName)
+            .singleActiveConsumer()
+            .messageHandler(
+                (context, message) -> {
+                  lastReceivedOffset.set(context.offset());
+                  receivedMessages.get(0).incrementAndGet();
+                })
+            .offset(OffsetSpecification.first())
+            .manualTrackingStrategy()
+            .builder()
+            .consumerUpdateListener(context -> OffsetSpecification.offset(lastReceivedOffset.get()))
+            .build();
+
+    Consumer consumer2 =
+        environment.consumerBuilder().stream(stream)
+            .name(consumerName)
+            .singleActiveConsumer()
+            .messageHandler(
+                (context, message) -> {
+                  lastReceivedOffset.set(context.offset());
+                  receivedMessages.get(1).incrementAndGet();
+                })
+            .offset(OffsetSpecification.first())
+            .manualTrackingStrategy()
+            .builder()
+            .consumerUpdateListener(context -> OffsetSpecification.offset(lastReceivedOffset.get()))
             .build();
 
     publishAndWaitForConfirms(cf, messageCount, stream);
@@ -101,5 +159,10 @@ void autoTrackingSecondConsumerShouldTakeOverWhereTheFirstOneLeftOff() throws Ex
     waitAtMost(() -> receivedMessages.getOrDefault(0, new AtomicInteger(1)).get() == messageCount);
     assertThat(lastReceivedOffset).hasValueGreaterThan(firstWaveLimit);
     assertThat(receivedMessages.get(0)).hasValue(messageCount);
+
+    consumer2.close();
+
+    // nothing stored on the server side
+    assertThat(cf.get().queryOffset(consumerName, stream)).isZero();
   }
 }