Skip to content

Reduce work in testBulkIndexingRequestSplitting #145188

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
DaveCTurner merged 2 commits into from
Mar 30, 2026
+10 −27
Merged

Reduce work in testBulkIndexingRequestSplitting #145188

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
01cd0e1
Reduce work in `testBulkIndexingRequestSplitting`
DaveCTurner Mar 30, 2026
807f0a8
Revert test logging
DaveCTurner Mar 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 10 additions & 27 deletions ...nalClusterTest/java/org/elasticsearch/http/netty4/Netty4IncrementalRequestHandlingIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,6 @@
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;

Expand All @@ -49,7 +46,6 @@
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;
Expand Down Expand Up @@ -77,7 +73,6 @@
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;
Expand Down Expand Up @@ -528,13 +523,8 @@ 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 watermarkBytes = between(100, 2000);
final var tinyNode = internalCluster().startCoordinatingOnlyNode(
Settings.builder()
.put(IndexingPressure.SPLIT_BULK_LOW_WATERMARK.getKey(), ByteSizeValue.ofBytes(watermarkBytes))
Expand All @@ -551,14 +541,6 @@ public void testBulkIndexingRequestSplitting() throws Exception {
channel.writeAndFlush(request);

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(),
Expand All @@ -570,24 +552,28 @@ public void testBulkIndexingRequestSplitting() throws Exception {

final var valueLength = between(10, 30);
final var docSizeBytes = "{'field':''}".length() + valueLength;
final var itemCount = between(watermarkBytes / docSizeBytes + 1, 300); // enough to split at least once
final var minItemCount = watermarkBytes / docSizeBytes + 1;
final var itemCount = between(minItemCount, minItemCount * 2); // enough to split at least once
assertThat(itemCount * docSizeBytes, greaterThan(watermarkBytes));
for (int i = 0; i < itemCount; i++) {
channel.write(new DefaultHttpContent(Unpooled.wrappedBuffer(Strings.format("""
{"index":{"_index":"%s"}}
{"field":"%s"}
""", indexName, randomAlphaOfLength(valueLength)).getBytes(StandardCharsets.UTF_8))));

if (i == minItemCount && randomBoolean()) {
channel.flush();
if (randomBoolean()) {
safeAwait(indexCreatedListener);
}
}
}

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;
Expand Down Expand Up @@ -667,9 +653,6 @@ 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>() {
Expand Down
Loading