You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pt...@apache.org on 2023/12/15 16:08:46 UTC
(ignite-3) branch main updated: IGNITE-21083 Fix IllegalReferenceCountException in JDBC and Java client (#2961)
This is an automated email from the ASF dual-hosted git repository.
ptupitsyn pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 2efc454f21 IGNITE-21083 Fix IllegalReferenceCountException in JDBC and Java client (#2961)
2efc454f21 is described below
commit 2efc454f21460ec31a7828d8927ed113d0c80655
Author: Pavel Tupitsyn <pt...@apache.org>
AuthorDate: Fri Dec 15 18:08:40 2023 +0200
IGNITE-21083 Fix IllegalReferenceCountException in JDBC and Java client (#2961)
**Problem:** `payloadReader` can escape `asyncContinuationExecutor` thread in some cases due to `thenApply` call in `TcpClientChannel.receiveAsync`. In particular, this happened due to blocking future wait in `JdbcResultSet.close0`. In this case, `TcpClientChannel.onMessage` closes the buffer before we read from it, causing `IllegalReferenceCountException`.
**Solution:** ensure that `payloadReader` is called synchronously before we close the buffer. Store it separately and invoke directly without relying on future continuations.
Note: `supportedVers` and related handshake retry removed for simplicity - we only have one supported protocol version currently.
---
.../ignite/internal/client/TcpClientChannel.java | 91 +++++++++-------------
1 file changed, 36 insertions(+), 55 deletions(-)
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
index 6be52a695a..30e5cceb2d 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.client;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.ExceptionUtils.copyExceptionWithCause;
import static org.apache.ignite.internal.util.ExceptionUtils.sneakyThrow;
import static org.apache.ignite.lang.ErrorGroups.Client.CONNECTION_ERR;
@@ -28,7 +27,6 @@ import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import java.net.InetSocketAddress;
import java.util.Collection;
-import java.util.Collections;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
@@ -70,11 +68,6 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
/** Protocol version used by default on first connection attempt. */
private static final ProtocolVersion DEFAULT_VERSION = ProtocolVersion.LATEST_VER;
- /** Supported protocol versions. */
- private static final Collection<ProtocolVersion> supportedVers = Collections.singletonList(
- ProtocolVersion.V3_0_0
- );
-
/** Minimum supported heartbeat interval. */
private static final long MIN_RECOMMENDED_HEARTBEAT_INTERVAL = 500;
@@ -94,7 +87,7 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
private final AtomicLong reqId = new AtomicLong(1);
/** Pending requests. */
- private final Map<Long, ClientRequestFuture> pendingReqs = new ConcurrentHashMap<>();
+ private final Map<Long, ClientRequestFuture<?>> pendingReqs = new ConcurrentHashMap<>();
/** Notification handlers. */
private final Map<Long, NotificationHandler> notificationHandlers = new ConcurrentHashMap<>();
@@ -219,7 +212,7 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
sock.close();
}
- for (ClientRequestFuture pendingReq : pendingReqs.values()) {
+ for (ClientRequestFuture<?> pendingReq : pendingReqs.values()) {
pendingReq.completeExceptionally(new IgniteClientConnectionException(CONNECTION_ERR, "Channel is closed", cause));
}
@@ -237,12 +230,10 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
@Override
public void onMessage(ByteBuf buf) {
asyncContinuationExecutor.execute(() -> {
- try {
- processNextMessage(buf);
+ try (var unpacker = new ClientMessageUnpacker(buf)) {
+ processNextMessage(unpacker);
} catch (Throwable t) {
close(t, false);
- } finally {
- buf.release();
}
});
}
@@ -278,14 +269,9 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
notificationHandlers.put(id, notificationHandler);
}
- ClientRequestFuture fut = send(opCode, id, payloadWriter);
-
- return receiveAsync(fut, payloadReader);
+ return send(opCode, id, payloadWriter, payloadReader);
} catch (Throwable t) {
- CompletableFuture<T> fut = new CompletableFuture<>();
- fut.completeExceptionally(t);
-
- return fut;
+ return CompletableFuture.failedFuture(t);
}
}
@@ -297,12 +283,13 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
* @param payloadWriter Payload writer to stream or {@code null} if request has no payload.
* @return Request future.
*/
- private ClientRequestFuture send(int opCode, long id, @Nullable PayloadWriter payloadWriter) {
+ private <T> ClientRequestFuture<T> send(
+ int opCode, long id, @Nullable PayloadWriter payloadWriter, @Nullable PayloadReader<T> payloadReader) {
if (closed()) {
throw new IgniteClientConnectionException(CONNECTION_ERR, "Channel is closed");
}
- ClientRequestFuture fut = new ClientRequestFuture();
+ ClientRequestFuture<T> fut = new ClientRequestFuture<>(payloadReader);
pendingReqs.put(id, fut);
metrics.requestsActiveIncrement();
@@ -352,37 +339,33 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
}
/**
- * Receives the response asynchronously.
+ * Completes the request future.
*
* @param pendingReq Request future.
- * @param payloadReader Payload reader from stream.
- * @return Future for the operation.
*/
- private <T> CompletableFuture<T> receiveAsync(ClientRequestFuture pendingReq, @Nullable PayloadReader<T> payloadReader) {
- return pendingReq.thenApply(payload -> {
- if (payload == null || payloadReader == null) {
- return null;
- }
-
+ private <T> void complete(ClientRequestFuture<T> pendingReq, ClientMessageUnpacker unpacker) {
+ if (pendingReq.payloadReader == null) {
+ pendingReq.complete(null);
+ } else {
try {
- return payloadReader.apply(new PayloadInputChannel(this, payload));
+ T res = pendingReq.payloadReader.apply(new PayloadInputChannel(this, unpacker));
+ pendingReq.complete(res);
} catch (Exception e) {
log.error("Failed to deserialize server response [remoteAddress=" + cfg.getAddress() + "]: " + e.getMessage(), e);
- throw new IgniteException(PROTOCOL_ERR, "Failed to deserialize server response: " + e.getMessage(), e);
+ pendingReq.completeExceptionally(
+ new IgniteException(PROTOCOL_ERR, "Failed to deserialize server response: " + e.getMessage(), e));
}
- });
+ }
}
/**
* Process next message from the input stream and complete corresponding future.
*/
- private void processNextMessage(ByteBuf buf) throws IgniteException {
- var unpacker = new ClientMessageUnpacker(buf);
-
+ private void processNextMessage(ClientMessageUnpacker unpacker) throws IgniteException {
if (protocolCtx == null) {
// Process handshake.
- pendingReqs.remove(-1L).complete(unpacker);
+ complete(pendingReqs.remove(-1L), unpacker);
return;
}
@@ -399,7 +382,7 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
return;
}
- ClientRequestFuture pendingReq = pendingReqs.remove(resId);
+ ClientRequestFuture<?> pendingReq = pendingReqs.remove(resId);
if (pendingReq == null) {
log.error("Unexpected response ID [remoteAddress=" + cfg.getAddress() + "]: " + resId);
@@ -410,7 +393,7 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
if (err == null) {
metrics.requestsCompletedIncrement();
- pendingReq.complete(unpacker);
+ complete(pendingReq, unpacker);
} else {
metrics.requestsFailedIncrement();
notificationHandlers.remove(resId);
@@ -548,9 +531,9 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
}
/** Client handshake. */
- private CompletableFuture<Void> handshakeAsync(ProtocolVersion ver)
+ private CompletableFuture<Object> handshakeAsync(ProtocolVersion ver)
throws IgniteClientConnectionException {
- ClientRequestFuture fut = new ClientRequestFuture();
+ ClientRequestFuture<Object> fut = new ClientRequestFuture<>(r -> handshakeRes(r.in()));
pendingReqs.put(-1L, fut);
handshakeReqAsync(ver).addListener(f -> {
@@ -565,7 +548,6 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
}
return fut
- .thenCompose(res -> handshakeRes(res, ver))
.handle((res, err) -> {
if (err != null) {
if (err instanceof TimeoutException || err.getCause() instanceof TimeoutException) {
@@ -621,18 +603,11 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
return write(req);
}
- /** Receive and handle handshake response. */
- private CompletableFuture<Void> handshakeRes(ClientMessageUnpacker unpacker, ProtocolVersion proposedVer) {
+ private @Nullable Object handshakeRes(ClientMessageUnpacker unpacker) {
try {
- ProtocolVersion srvVer = new ProtocolVersion(unpacker.unpackShort(), unpacker.unpackShort(),
- unpacker.unpackShort());
+ ProtocolVersion srvVer = new ProtocolVersion(unpacker.unpackShort(), unpacker.unpackShort(), unpacker.unpackShort());
if (!unpacker.tryUnpackNil()) {
- if (!proposedVer.equals(srvVer) && supportedVers.contains(srvVer)) {
- // Retry with server version.
- return handshakeAsync(srvVer);
- }
-
throw sneakyThrow(readError(unpacker));
}
@@ -652,11 +627,11 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
protocolCtx = new ProtocolContext(
srvVer, ProtocolBitmaskFeature.allFeaturesAsEnumSet(), serverIdleTimeout, clusterNode, clusterId);
- return nullCompletedFuture();
+ return null;
} catch (Exception e) {
log.warn("Failed to handle handshake response [remoteAddress=" + cfg.getAddress() + "]: " + e.getMessage(), e);
- return CompletableFuture.failedFuture(e);
+ throw e;
}
}
@@ -725,7 +700,13 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
/**
* Client request future.
*/
- private static class ClientRequestFuture extends CompletableFuture<ClientMessageUnpacker> {
+ private static class ClientRequestFuture<T> extends CompletableFuture<T> {
+ @Nullable
+ private final PayloadReader<T> payloadReader;
+
+ private ClientRequestFuture(@Nullable PayloadReader<T> payloadReader) {
+ this.payloadReader = payloadReader;
+ }
}
/**