Use server-side offset tracking in SAC test

Instead of client-side in-memory tracking, because it's more realistic.

References rabbitmq/rabbitmq-server#3753

Author: acogoluegnes

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

@@ -38,13 +38,15 @@ public class SingleActiveConsumerTest {
 
   @Test
   void secondSubscriptionShouldTakeOverAfterFirstOneUnsubscribes() throws Exception {
+    Client writerClient = cf.get();
     int messageCount = 10000;
     AtomicLong lastReceivedOffset = new AtomicLong(0);
     Map<Byte, Boolean> consumerStates = new ConcurrentHashMap<>();
     Map<Byte, AtomicInteger> receivedMessages = new ConcurrentHashMap<>();
     receivedMessages.put(b(0), new AtomicInteger(0));
     receivedMessages.put(b(1), new AtomicInteger(0));
     CountDownLatch consumerUpdateLatch = new CountDownLatch(2);
+    String consumerName = "foo";
     ClientParameters clientParameters =
         new ClientParameters()
             .chunkListener(
@@ -59,17 +61,17 @@ void secondSubscriptionShouldTakeOverAfterFirstOneUnsubscribes() throws Exceptio
                 (client, subscriptionId, active) -> {
                   consumerStates.put(subscriptionId, active);
                   consumerUpdateLatch.countDown();
-                  if (lastReceivedOffset.get() == 0) {
+                  long storedOffset = writerClient.queryOffset(consumerName, stream);
+                  if (storedOffset == 0) {
                     return OffsetSpecification.first();
                   } else {
-                    return OffsetSpecification.offset(lastReceivedOffset.get() + 1);
+                    return OffsetSpecification.offset(storedOffset + 1);
                   }
                 });
     Client client = cf.get(clientParameters);
 
     TestUtils.publishAndWaitForConfirms(cf, messageCount, stream);
 
-    String consumerName = "foo";
     Map<String, String> parameters = new HashMap<>();
     parameters.put("single-active-consumer", "true");
     parameters.put("name", consumerName);
@@ -87,8 +89,8 @@ void secondSubscriptionShouldTakeOverAfterFirstOneUnsubscribes() throws Exceptio
         () -> receivedMessages.getOrDefault(b(0), new AtomicInteger(0)).get() == messageCount);
 
     assertThat(lastReceivedOffset).hasPositiveValue();
-    client.storeOffset(consumerName, stream, lastReceivedOffset.get());
-    waitAtMost(() -> client.queryOffset(consumerName, stream) == lastReceivedOffset.get());
+    writerClient.storeOffset(consumerName, stream, lastReceivedOffset.get());
+    waitAtMost(() -> writerClient.queryOffset(consumerName, stream) == lastReceivedOffset.get());
 
     long firstWaveLimit = lastReceivedOffset.get();
     response = client.unsubscribe(b(0));