Fix correlation ID increment

Fixes #689

Author: acogoluegnes

Diff for: src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -104,6 +104,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 import java.util.function.ToLongFunction;
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLHandshakeException;
@@ -178,7 +179,7 @@ public long applyAsLong(Object value) {
   private final CredentialsProvider credentialsProvider;
   private final Runnable nettyClosing;
   private final int maxFrameSize;
-  private final boolean frameSizeCopped;
+  private final boolean frameSizeCapped;
   private final EventLoopGroup eventLoopGroup;
   private final Map<String, String> clientProperties;
   private final String NETTY_HANDLER_FLUSH_CONSOLIDATION =
@@ -373,12 +374,18 @@ public void initChannel(SocketChannel ch) {
           new TuneState(
               parameters.requestedMaxFrameSize, (int) parameters.requestedHeartbeat.getSeconds());
       this.clientProperties = clientProperties(parameters.clientProperties);
+      debug(() -> "exchanging peer properties");
       this.serverProperties = peerProperties();
+      debug(() -> "peer properties exchanged");
+      debug(() -> "starting SASL handshake");
       this.saslMechanisms = getSaslMechanisms();
+      debug(() -> "SASL mechanisms supported by server ({})", this.saslMechanisms);
+      debug(() -> "starting authentication");
       authenticate(this.credentialsProvider);
+      debug(() -> "authenticated");
       this.tuneState.await(Duration.ofSeconds(10));
       this.maxFrameSize = this.tuneState.getMaxFrameSize();
-      this.frameSizeCopped = this.maxFrameSize() > 0;
+      this.frameSizeCapped = this.maxFrameSize() > 0;
       LOGGER.debug(
           "Connection tuned with max frame size {} and heartbeat {}",
           this.maxFrameSize(),
@@ -418,6 +425,8 @@ public void initChannel(SocketChannel ch) {
       started.set(true);
       this.metricsCollector.openConnection();
     } catch (RuntimeException e) {
+      LOGGER.debug(
+          "Error while opening connection {}: {}", this.clientConnectionName, e.getMessage());
       this.closingSequence(null);
       throw e;
     }
@@ -462,10 +471,14 @@ int maxFrameSize() {
     return this.maxFrameSize;
   }
 
+  private int nextCorrelationId() {
+    return this.correlationSequence.getAndIncrement();
+  }
+
   private Map<String, String> peerProperties() {
     int clientPropertiesSize = mapSize(this.clientProperties);
     int length = 2 + 2 + 4 + clientPropertiesSize;
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocateNoCheck(length + 4);
       bb.writeInt(length);
@@ -474,6 +487,7 @@ private Map<String, String> peerProperties() {
       bb.writeInt(correlationId);
       writeMap(bb, this.clientProperties);
       OutstandingRequest<Map<String, String>> request = outstandingRequest();
+      LOGGER.debug("Peer properties request has correlation ID {}", correlationId);
       outstandingRequests.put(correlationId, request);
       channel.writeAndFlush(bb);
       request.block();
@@ -539,7 +553,7 @@ private SaslAuthenticateResponse sendSaslAuthenticate(
             + saslMechanism.getName().length()
             + 4
             + (challengeResponse == null ? 0 : challengeResponse.length);
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocateNoCheck(length + 4);
       bb.writeInt(length);
@@ -569,7 +583,7 @@ private SaslAuthenticateResponse sendSaslAuthenticate(
 
   private Map<String, String> open(String virtualHost) {
     int length = 2 + 2 + 4 + 2 + virtualHost.length();
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -610,7 +624,7 @@ void send(byte[] content) {
 
   private void sendClose(short code, String reason) {
     int length = 2 + 2 + 4 + 2 + 2 + reason.length();
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -643,7 +657,7 @@ private void sendClose(short code, String reason) {
 
   private List<String> getSaslMechanisms() {
     int length = 2 + 2 + 4;
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocateNoCheck(length + 4);
       bb.writeInt(length);
@@ -670,7 +684,7 @@ public Response create(String stream) {
 
   public Response create(String stream, Map<String, String> arguments) {
     int length = 2 + 2 + 4 + 2 + stream.length() + mapSize(arguments);
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -719,7 +733,7 @@ Response createSuperStream(
             + collectionSize(partitions)
             + collectionSize(bindingKeys)
             + mapSize(arguments);
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -748,7 +762,7 @@ Response createSuperStream(
   Response deleteSuperStream(String superStream) {
     this.superStreamManagementCommandVersionsCheck.run();
     int length = 2 + 2 + 4 + 2 + superStream.length();
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -808,7 +822,7 @@ private static ByteBuf writeMap(ByteBuf bb, Map<String, String> elements) {
   }
 
   ByteBuf allocate(ByteBufAllocator allocator, int capacity) {
-    if (frameSizeCopped && capacity > this.maxFrameSize()) {
+    if (frameSizeCapped && capacity > this.maxFrameSize()) {
       throw new IllegalArgumentException(
           "Cannot allocate "
               + capacity
@@ -832,7 +846,7 @@ private ByteBuf allocateNoCheck(int capacity) {
 
   public Response delete(String stream) {
     int length = 2 + 2 + 4 + 2 + stream.length();
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -864,7 +878,7 @@ public Map<String, StreamMetadata> metadata(String... streams) {
       throw new IllegalArgumentException("At least one stream must be specified");
     }
     int length = 2 + 2 + 4 + arraySize(streams); // API code, version, correlation ID, array size
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -897,7 +911,7 @@ public Response declarePublisher(byte publisherId, String publisherReference, St
           "If specified, publisher reference must less than 256 characters");
     }
     int length = 2 + 2 + 4 + 1 + 2 + publisherReferenceSize + 2 + stream.length();
-    int correlationId = correlationSequence.getAndIncrement();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -928,7 +942,7 @@ public Response declarePublisher(byte publisherId, String publisherReference, St
 
   public Response deletePublisher(byte publisherId) {
     int length = 2 + 2 + 4 + 1;
-    int correlationId = correlationSequence.getAndIncrement();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -1252,7 +1266,7 @@ public Response subscribe(
       propertiesSize = mapSize(properties);
     }
     length += propertiesSize;
-    int correlationId = correlationSequence.getAndIncrement();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -1320,7 +1334,7 @@ public QueryOffsetResponse queryOffset(String reference, String stream) {
     }
 
     int length = 2 + 2 + 4 + 2 + reference.length() + 2 + stream.length();
-    int correlationId = correlationSequence.getAndIncrement();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -1361,7 +1375,7 @@ public long queryPublisherSequence(String publisherReference, String stream) {
     }
 
     int length = 2 + 2 + 4 + 2 + publisherReference.length() + 2 + stream.length();
-    int correlationId = correlationSequence.getAndIncrement();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -1398,7 +1412,7 @@ public long queryPublisherSequence(String publisherReference, String stream) {
 
   public Response unsubscribe(byte subscriptionId) {
     int length = 2 + 2 + 4 + 1;
-    int correlationId = correlationSequence.getAndIncrement();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -1445,6 +1459,8 @@ private void closeNetty() {
       if (this.channel != null && this.channel.isOpen()) {
         LOGGER.debug("Closing Netty channel");
         this.channel.close().get(10, TimeUnit.SECONDS);
+      } else {
+        LOGGER.debug("No Netty channel to close");
       }
     } catch (InterruptedException e) {
       LOGGER.info("Channel closing has been interrupted");
@@ -1530,7 +1546,7 @@ public List<String> route(String routingKey, String superStream) {
             + routingKey.length()
             + 2
             + superStream.length(); // API code, version, correlation ID, 2 strings
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -1565,7 +1581,7 @@ public List<String> partitions(String superStream) {
     }
     int length =
         2 + 2 + 4 + 2 + superStream.length(); // API code, version, correlation ID, 1 string
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -1593,7 +1609,7 @@ List<FrameHandlerInfo> exchangeCommandVersions() {
     List<FrameHandlerInfo> commandVersions = ServerFrameHandler.commandVersions();
     int length = 2 + 2 + 4 + 4; // API code, version, correlation ID, array size
     length += commandVersions.size() * (2 + 2 + 2);
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -1626,7 +1642,7 @@ StreamStatsResponse streamStats(String stream) {
       throw new IllegalArgumentException("stream must not be null");
     }
     int length = 2 + 2 + 4 + 2 + stream.length(); // API code, version, correlation ID, 1 string
-    int correlationId = correlationSequence.incrementAndGet();
+    int correlationId = nextCorrelationId();
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
@@ -2583,6 +2599,10 @@ public ClientParameters bootstrapCustomizer(Consumer<Bootstrap> bootstrapCustomi
       return this;
     }
 
+    Duration rpcTimeout() {
+      return this.rpcTimeout;
+    }
+
     ClientParameters duplicate() {
       ClientParameters duplicate = new ClientParameters();
       for (Field field : ClientParameters.class.getDeclaredFields()) {
@@ -2926,4 +2946,10 @@ private <T> OutstandingRequest<T> outstandingRequest() {
   public String toString() {
     return "Client{connectionName='" + connectionName() + "'}";
   }
+
+  private void debug(Supplier<String> format, Object... args) {
+    if (LOGGER.isDebugEnabled()) {
+      LOGGER.debug("Connection '" + this.clientConnectionName + "': " + format.get(), args);
+    }
+  }
 }