Skip to content
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

fix: Block writer thread if response output buffer is full #5386

Merged
merged 3 commits into from
May 19, 2020
Merged

fix: Block writer thread if response output buffer is full #5386

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4223187
fix: block on write if write queue full
May 18, 2020
a3982d6
use cf for blocking
May 18, 2020
c0b0654
tweaks
May 18, 2020
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
22 changes: 21 additions & 1 deletion ksqldb-rest-app/src/main/java/io/confluent/ksql/api/server/ResponseOutputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,14 @@

package io.confluent.ksql.api.server;

import io.confluent.ksql.util.KsqlException;
import io.confluent.ksql.util.VertxUtils;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpServerResponse;
import java.io.OutputStream;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.jetbrains.annotations.NotNull;

/*
Expand All @@ -41,7 +45,7 @@ public void write(final int b) {
}

@Override
public void write(final @NotNull byte[] bytes, final int offset, final int length) {
public synchronized void write(final @NotNull byte[] bytes, final int offset, final int length) {
Copy link
Contributor

Choose a reason for hiding this comment

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

ResponseOutputStream is only written to by a single thread anyway, right? (Trying to check my understanding.)

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes, but it's a sanity check.

Objects.requireNonNull(bytes);
if ((offset < 0) || (offset > bytes.length)) {
throw new IndexOutOfBoundsException();
Expand All @@ -55,13 +59,29 @@ public void write(final @NotNull byte[] bytes, final int offset, final int lengt
final byte[] bytesToWrite = new byte[length];
System.arraycopy(bytes, offset, bytesToWrite, 0, length);
final Buffer buffer = Buffer.buffer(bytesToWrite);
blockIfWriteQueueFull();
response.write(buffer);
}

@Override
public void close() {
response.end();
}

private void blockIfWriteQueueFull() {
VertxUtils.checkIsWorker();
if (response.writeQueueFull()) {
final CompletableFuture<Void> cf = new CompletableFuture<>();
response.drainHandler(v -> cf.complete(null));
try {
cf.get(60, TimeUnit.SECONDS);
} catch (Exception e) {
Copy link
Contributor

Choose a reason for hiding this comment

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

Should we check explicitly for TimeoutException? Since we don't expect ExecutionException to be thrown here.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

We could, but I don't think it adds much as we want to rethrow on any exception we receive.

// Very slow consumers will result in a timeout, this will cause the push query to be closed
throw new KsqlException(e);
Copy link
Contributor

Choose a reason for hiding this comment

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

Should we add a more informative error message?

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 wrapped exception should contain the important error messages. I don't usually like adding new messages on rethrows unless you're doing something extra.

}
}
}

}


2 changes: 2 additions & 0 deletions ksqldb-rest-app/src/main/java/io/confluent/ksql/api/server/ServerVerticle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,8 @@ private void handleQueryRequest(final RoutingContext routingContext) {
final CompletableFuture<Void> connectionClosedFuture = new CompletableFuture<>();
routingContext.request().connection().closeHandler(v -> connectionClosedFuture.complete(null));

routingContext.response().setWriteQueueMaxSize(200 * 1024);

handleOldApiRequest(server, routingContext, KsqlRequest.class,
(request, apiSecurityContext) ->
endpoints
Expand Down