Skip to content

WIP: Increase http client's max chunk size #10071

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

Draft
wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

WIP: Increase http client's max chunk size #10071

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c09611f
WIP: Increase http client's max chunk size
Gerrrr Sep 21, 2023
3758223
Also set max chunk size on the server
lucasbru Sep 22, 2023
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
4 changes: 3 additions & 1 deletion ksqldb-rest-app/src/main/java/io/confluent/ksql/api/server/Server.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ public class Server {
private URI internalListener;
private WorkerExecutor workerExecutor;
private FileWatcher fileWatcher;
static final int CHUNK_SIZE = 16 * 1024;

@SuppressFBWarnings(value = "EI_EXPOSE_REP2")
public Server(
Expand Down Expand Up @@ -318,7 +319,8 @@ private static HttpServerOptions createHttpServerOptions(final KsqlRestConfig ks
.setIdleTimeout(idleTimeoutSeconds).setIdleTimeoutUnit(TimeUnit.SECONDS)
.setPerMessageWebSocketCompressionSupported(true)
.setPerFrameWebSocketCompressionSupported(true)
.setUseProxyProtocol(isProxyProtocolListener);
.setUseProxyProtocol(isProxyProtocolListener)
.setMaxChunkSize(CHUNK_SIZE);

if (tls) {
final String ksConfigName = isInternalListener
Expand Down
4 changes: 3 additions & 1 deletion ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/PreconditionServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ public class PreconditionServer {
private final List<URI> listeners;
private final Set<URI> proxyProtocolListeners;
private FileWatcher fileWatcher;
static final int CHUNK_SIZE = 16 * 1024;

@SuppressFBWarnings(value = "EI_EXPOSE_REP2")
public PreconditionServer(
Expand Down Expand Up @@ -167,7 +168,8 @@ private static HttpServerOptions createHttpServerOptions(
.setIdleTimeout(idleTimeoutSeconds).setIdleTimeoutUnit(TimeUnit.SECONDS)
.setPerMessageWebSocketCompressionSupported(true)
.setPerFrameWebSocketCompressionSupported(true)
.setUseProxyProtocol(useProxyProtocol);
.setUseProxyProtocol(useProxyProtocol)
.setMaxChunkSize(CHUNK_SIZE);

if (tls) {
final String ksConfigName = KsqlRestConfig.KSQL_SSL_KEYSTORE_ALIAS_EXTERNAL_CONFIG;
Expand Down
8 changes: 6 additions & 2 deletions ksqldb-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlRestClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,8 @@ public final class KsqlRestClient implements Closeable {
static final String CCLOUD_CONNECT_USERNAME_HEADER = "X-Confluent-API-Key";
static final String CCLOUD_CONNECT_PASSWORD_HEADER = "X-Confluent-API-Secret";

static final int CHUNK_SIZE = 16 * 1024;

private final KsqlClient client;
private final LocalProperties localProperties;
private final AtomicReference<String> serializedConsistencyVector;
Expand Down Expand Up @@ -113,8 +115,10 @@ public static KsqlRestClient create(
creds,
ccloudApiKey,
(cprops, credz, lprops) -> new KsqlClient(cprops, credz, lprops,
new HttpClientOptions(),
Optional.of(new HttpClientOptions().setProtocolVersion(HttpVersion.HTTP_2)))
new HttpClientOptions().setMaxChunkSize(CHUNK_SIZE),
Optional.of(new HttpClientOptions()
.setProtocolVersion(HttpVersion.HTTP_2)
.setMaxChunkSize(CHUNK_SIZE)))
Comment on lines +118 to +121
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are you sure you need to increase the chunk size on the client and not on the server?
The post this PR is based on mentions the server. Are these options propagated?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These are client options, not server options. We need to make the change to the server I believe. Not sure what happens if it's out of agreement with the client.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The Google Group post described an issue on the server side. Check the classes at bottom of the stack trace:

java.lang.IndexOutOfBoundsException: writerIndex: 10614 (expected: readerIndex(0) <= writerIndex <= capacity(8192))
        at io.netty.buffer.AbstractByteBuf.writerIndex(AbstractByteBuf.java:118)
        at io.netty.buffer.CompositeByteBuf.writerIndex(CompositeByteBuf.java:1686)
        at io.vertx.core.http.impl.HttpClientRequestImpl.write(HttpClientRequestImpl.java:851)
        at io.vertx.core.http.impl.HttpClientRequestImpl.write(HttpClientRequestImpl.java:228)
        at io.vertx.core.http.impl.HttpClientRequestImpl.write(HttpClientRequestImpl.java:51)
        at io.vertx.core.streams.impl.PumpImpl.lambda$new$1(PumpImpl.java:64)
        at io.vertx.core.http.impl.HttpServerRequestImpl.handleData(HttpServerRequestImpl.java:373)
        at io.vertx.core.http.impl.ServerConnection.handleChunk(ServerConnection.java:293)
        at io.vertx.core.http.impl.ServerConnection.processMessage(ServerConnection.java:435)
        at io.vertx.core.http.impl.ServerConnection.handleMessage(ServerConnection.java:131)**
        at io.vertx.core.http.impl.HttpServerImpl$ServerHandler.doMessageReceived(HttpServerImpl.java:678)
        at io.vertx.core.http.impl.HttpServerImpl$ServerHandler.doMessageReceived(HttpServerImpl.java:573)

while the stack trace we observed happens on the client side. Check the classes at bottom of the stack trace:

java.lang.IndexOutOfBoundsException: writerIndex(8192) + minWritableBytes(7569) exceeds maxCapacity(8192): UnpooledSlicedByteBuf(ridx: 0, widx: 8192, cap: 8192/8192, unwrapped: UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 16384, widx: 16384, cap: 16413))
	at io.netty.buffer.AbstractByteBuf.ensureWritable0(AbstractByteBuf.java:294)
	at io.netty.buffer.AbstractByteBuf.ensureWritable(AbstractByteBuf.java:280)
	at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1103)
	at io.vertx.core.buffer.impl.BufferImpl.appendBuffer(BufferImpl.java:268)
	at io.vertx.core.parsetools.impl.RecordParserImpl.handle(RecordParserImpl.java:283)
	at io.vertx.core.parsetools.impl.RecordParserImpl.handle(RecordParserImpl.java:27)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:264)
	at io.vertx.core.http.impl.HttpEventHandler.handleChunk(HttpEventHandler.java:51)
	at io.vertx.core.http.impl.HttpClientResponseImpl.handleChunk(HttpClientResponseImpl.java:239)
	at io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.lambda$new$0(Http1xClientConnection.java:443)
	at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:239)
	at io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:129)
	at io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.handleChunk(Http1xClientConnection.java:678)

I thought that these are 2 similar issues, but happening in different places; hence, changed configuration on the client side.

I agree with @AlanConfluent that we should also change the chunk size on the server side to make client and server config match.

);
}

Expand Down