Log Netty channel writability history

Author: acogoluegnes

.github/workflows/test.yml

@@ -46,15 +46,16 @@ jobs:
             -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
             -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
             -Dmaven.javadoc.skip=true \
-            --no-transfer-progress
-      - name: Test with blocking IO
-        run: |
-          ./mvnw verify -Dio.layer=socket -Drabbitmqctl.bin=DOCKER:rabbitmq0 \
-            -Dtest-broker.A.nodename=rabbit@node0 -Dtest-broker.B.nodename=rabbit@node1 \
-            -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
-            -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
-            -Dmaven.javadoc.skip=true \
-            --no-transfer-progress
+            --no-transfer-progress \
+            -Dit.test=Confirm#persistentMandatoryCombinations,RequeueOnConnectionClose#requeueInFlightConsumerNoAck
+      # - name: Test with blocking IO
+      #   run: |
+      #     ./mvnw verify -Dio.layer=socket -Drabbitmqctl.bin=DOCKER:rabbitmq0 \
+      #       -Dtest-broker.A.nodename=rabbit@node0 -Dtest-broker.B.nodename=rabbit@node1 \
+      #       -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
+      #       -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
+      #       -Dmaven.javadoc.skip=true \
+      #       --no-transfer-progress
       - name: Stop cluster
         run: docker compose --file ci/cluster/docker-compose.yml down
       - name: Publish snapshot

src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java

@@ -51,6 +51,8 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.time.Duration;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -322,6 +324,7 @@ public void writeFrame(Frame frame) throws IOException {
             if (canWriteNow) {
               this.doWriteFrame(frame);
             } else {
+              this.handler.logEvents();
               throw new IOException("Frame enqueuing failed");
             }
           } catch (InterruptedException e) {
@@ -345,6 +348,7 @@ public void flush() {
     @Override
     public void close() {
       if (this.closed.compareAndSet(false, true)) {
+        this.handler.logEvents();
         Runnable closing = () -> closeNettyState(this.channel, this.eventLoopGroup);
         if (this.channel.eventLoop().inEventLoop()) {
           this.channel.eventLoop().submit(closing);
@@ -438,9 +442,43 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
       }
     }
 
+    private static class Event {
+
+      private final long time;
+      private final String label;
+
+      public Event(long time, String label) {
+        this.time = time;
+        this.label = label;
+      }
+
+      @Override
+      public String toString() {
+        return this.label + " " + this.time;
+      }
+    }
+
+    private static final int MAX_EVENTS = 100;
+    private final Queue<Event> events = new ConcurrentLinkedQueue<>();
+
+    private void logEvents() {
+      if (this.events.size() > 0) {
+        long start = this.events.peek().time;
+        LOGGER.info("channel writability history:");
+        events.forEach(e -> LOGGER.info("{}: {}", (e.time - start) / 1_000_000, e.label));
+      }
+    }
+
     @Override
     public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
       boolean canWrite = ctx.channel().isWritable();
+      Event event = new Event(System.nanoTime(), String.valueOf(canWrite));
+      if (this.events.size() >= MAX_EVENTS) {
+        this.events.poll();
+        this.events.offer(event);
+      }
+      this.events.add(event);
+
       if (this.writable.compareAndSet(!canWrite, canWrite)) {
         if (canWrite) {
           CountDownLatch latch = writableLatch.getAndSet(new CountDownLatch(1));

src/test/resources/logback-test.xml

@@ -8,6 +8,7 @@
     <logger name="com.rabbitmq.client.AmqpClientTestExtension" level="debug" />
     <logger name="com.rabbitmq.client.test.server.TopicPermissions" level="debug" />
     <logger name="com.rabbitmq.client.impl.AbstractMetricsCollector" level="debug" />
+    <logger name="com.rabbitmq.client.impl.NettyFrameHandlerFactory" level="info" />
 
     <root level="warn">
         <appender-ref ref="STDOUT" />