Investigate testBulkIndexingRequestSplitting failure

modules/transport-netty4/src/internalClusterTest/java/org/elasticsearch/http/netty4/Netty4IncrementalRequestHandlingIT.java

@@ -34,6 +34,9 @@
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpUtil;
 import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.handler.logging.ByteBufFormat;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
 import io.netty.handler.stream.ChunkedStream;
 import io.netty.handler.stream.ChunkedWriteHandler;
 
@@ -46,6 +49,7 @@
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.collect.Iterators;
@@ -73,6 +77,7 @@
 import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.MockLog;
+import org.elasticsearch.test.junit.annotations.TestIssueLogging;
 import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.test.rest.ObjectPath;
 import org.elasticsearch.transport.Transports;
@@ -523,6 +528,11 @@ private void assertHttpBodyLogging(Consumer<ClientContext> test) throws Exceptio
         }
     }
 
+    @TestIssueLogging(
+        issueUrl = "https://github.com/elastic/elasticsearch/issues/144579",
+        value = "org.elasticsearch.http.netty4.Netty4IncrementalRequestHandlingIT:DEBUG"
+            + ",org.elasticsearch.transport.TransportService.tracer:TRACE"
+    )
     public void testBulkIndexingRequestSplitting() throws Exception {
         final var watermarkBytes = between(100, 200);
         final var tinyNode = internalCluster().startCoordinatingOnlyNode(
@@ -540,7 +550,15 @@ public void testBulkIndexingRequestSplitting() throws Exception {
             final var channel = clientContext.channel();
             channel.writeAndFlush(request);
 
-            final var indexName = randomIdentifier();
+            final var indexName = randomIndexName();
+            final var clusterStateLoggingListener = ClusterServiceUtils.addTemporaryStateListener(
+                internalCluster().getCurrentMasterNodeInstance(ClusterService.class),
+                cs -> {
+                    logger.info("cluster state: {}", cs);
+                    return false;
+                },
+                TimeValue.ONE_HOUR
+            );
             final var indexCreatedListener = ClusterServiceUtils.addTemporaryStateListener(
                 cs -> Iterators.filter(
                     cs.metadata().indicesAllProjects().iterator(),
@@ -564,8 +582,12 @@ public void testBulkIndexingRequestSplitting() throws Exception {
             channel.flush();
             safeAwait(indexCreatedListener); // index must be created before we finish sending the request
 
+            logger.info("--> completing request");
             channel.writeAndFlush(new DefaultLastHttpContent());
+            logger.info("--> awaiting response");
             final var response = clientContext.getNextResponse();
+            logger.info("--> received response");
+            clusterStateLoggingListener.onResponse(null);
             try {
                 assertEquals(RestStatus.OK.getStatus(), response.status().code());
                 final ObjectPath responseBody;
@@ -645,6 +667,9 @@ private ClientContext newClientContext(String nodeName, Consumer<Throwable> exce
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     var p = ch.pipeline();
+                    if (logger.isDebugEnabled()) {
+                        p.addLast(new LoggingHandler(Netty4IncrementalRequestHandlingIT.class, LogLevel.DEBUG, ByteBufFormat.HEX_DUMP));
+                    }
                     p.addLast(new HttpClientCodec());
                     p.addLast(new HttpObjectAggregator(ByteSizeUnit.MB.toIntBytes(4)));
                     p.addLast(new SimpleChannelInboundHandler<FullHttpResponse>() {

muted-tests.yml

@@ -306,9 +306,6 @@ tests:
 - class: org.elasticsearch.indices.IndicesRequestCacheIT
   method: testCanCache
   issue: https://github.com/elastic/elasticsearch/issues/144526
-- class: org.elasticsearch.http.netty4.Netty4IncrementalRequestHandlingIT
-  method: testBulkIndexingRequestSplitting
-  issue: https://github.com/elastic/elasticsearch/issues/144579
 - class: org.elasticsearch.xpack.esql.CsvIT
   method: test {csv-spec:spatial.convertCartesianFromStringParseError}
   issue: https://github.com/elastic/elasticsearch/issues/144580