diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java index f5da98ebd63..2788d0dc9e4 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java @@ -61,6 +61,7 @@ import org.apache.hadoop.hdds.scm.storage.DomainSocketFactory; import org.apache.hadoop.hdds.security.exception.SCMSecurityException; import org.apache.hadoop.hdds.tracing.TracingUtil; +import org.apache.hadoop.hdds.utils.IOUtils; import org.apache.hadoop.net.NetUtils; import org.apache.hadoop.net.unix.DomainSocket; import org.apache.hadoop.ozone.OzoneConfigKeys; @@ -246,6 +247,7 @@ private XceiverClientReply sendCommandWithTraceID( ContainerCommandResponseProto responseProto = null; IOException ioException = null; XceiverClientReply reply = new XceiverClientReply(null); + CompletableFuture future = null; if (request.getCmdType() != ContainerProtos.Type.GetBlock && request.getCmdType() != ContainerProtos.Type.Echo) { @@ -258,7 +260,8 @@ private XceiverClientReply sendCommandWithTraceID( LOG.debug("Executing command {} on datanode {}", request, dn); } reply.addDatanode(dn); - responseProto = sendCommandInternal(finalPayload).getResponse().get(); + future = sendCommandInternal(finalPayload).getResponse(); + responseProto = future.get(); if (validators != null && !validators.isEmpty()) { for (Validator validator : validators) { validator.accept(request, responseProto); @@ -287,6 +290,8 @@ private XceiverClientReply sendCommandWithTraceID( } catch (InterruptedException e) { LOG.error("Command execution was interrupted ", e); Thread.currentThread().interrupt(); + ioException = (IOException) new InterruptedIOException( + "Command " + processForDebug(request) + " was interrupted.").initCause(e); } if (responseProto != null) { @@ -294,6 +299,13 @@ private XceiverClientReply sendCommandWithTraceID( return reply; } else { Objects.requireNonNull(ioException); + if (future != null && request.getCmdType() == ContainerProtos.Type.GetBlock) { + // Complete the future before looking in the cache. If the stream has already arrived it is closed here; + // if it arrives later, the receiver thread finds the future completed and closes the stream itself. + future.completeExceptionally(ioException); + IOUtils.closeQuietly(getFileInputStream(finalPayload.getCallId(), + finalPayload.getGetBlock().getBlockID().getLocalID())); + } String message = "Failed to execute command {}"; if (LOG.isDebugEnabled()) { LOG.debug(message + " on the datanode {} {}.", request, dn, domainSocket, ioException); @@ -473,6 +485,7 @@ public void run() { long processStartTime = System.nanoTime(); ContainerProtos.Result result = responseProto.getResult(); if (result == ContainerProtos.Result.SUCCESS) { + BlockStreamKey streamKey = null; if (type == ContainerProtos.Type.GetBlock) { try { ContainerProtos.GetBlockResponseProto getBlockResponse = responseProto.getGetBlock(); @@ -496,7 +509,8 @@ public void run() { DATA_TRANSFER_MAGIC_CODE + ", Received: " + buf[0] + ")"); } DatanodeBlockID blockID = getBlockResponse.getBlockData().getBlockID(); - blockStreamCache.put(new BlockStreamKey(responseProto.getCallId(), blockID.getLocalID()), fis[0]); + streamKey = new BlockStreamKey(responseProto.getCallId(), blockID.getLocalID()); + blockStreamCache.put(streamKey, fis[0]); } catch (IOException e) { LOG.warn("Failed to handle short-circuit information exchange", e); // disable docket socket for a while @@ -505,7 +519,10 @@ public void run() { continue; } } - entry.getFuture().complete(responseProto); + if (!entry.getFuture().complete(responseProto) && streamKey != null) { + // the caller already gave up on this request, so nobody will claim the stream from the cache + IOUtils.closeQuietly(blockStreamCache.remove(streamKey)); + } } else { // response result is not SUCCESS entry.getFuture().complete(responseProto); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/TestXceiverServerDomainSocket.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/TestXceiverServerDomainSocket.java index d5e9b3c0eb1..31845f2e43e 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/TestXceiverServerDomainSocket.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/TestXceiverServerDomainSocket.java @@ -22,6 +22,7 @@ import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type.ReadChunk; import static org.apache.hadoop.hdds.scm.XceiverClientShortCircuit.vintPrefixed; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -38,6 +39,7 @@ import java.io.File; import java.io.FileInputStream; import java.io.IOException; +import java.io.InterruptedIOException; import java.net.ConnectException; import java.net.InetSocketAddress; import java.net.SocketException; @@ -46,15 +48,18 @@ import java.nio.channels.FileChannel; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.apache.commons.io.FileUtils; import org.apache.commons.io.IOUtils; import org.apache.commons.lang3.RandomStringUtils; +import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; @@ -62,9 +67,11 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; import org.apache.hadoop.hdds.scm.OzoneClientConfig; import org.apache.hadoop.hdds.scm.ScmConfigKeys; +import org.apache.hadoop.hdds.scm.XceiverClientShortCircuit; import org.apache.hadoop.hdds.scm.pipeline.MockPipeline; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.scm.storage.DomainSocketFactory; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; import org.apache.hadoop.net.unix.DomainSocket; import org.apache.hadoop.ozone.OzoneConfigKeys; import org.apache.hadoop.ozone.common.Checksum; @@ -130,6 +137,7 @@ public static void setup() { clientConfig.setShortCircuitReadDisableInterval(1000); DomainSocket.disableBindPathValidation(); conf.setFromObject(clientConfig); + DefaultMetricsSystem.setMiniClusterMode(true); metrics = ContainerMetrics.create(conf); readTimeout = 5 * 1000; writeTimeout = 5 * 1000; @@ -772,6 +780,76 @@ public void testGetBlockWithoutShortCircuitAccess() throws IOException { } } + /** + * The short-circuit client closes the block file stream of a GetBlock that its caller gave up on, + * whether the caller stops waiting before the response arrives or rejects the response after it arrived. + */ + @Test + public void testClientReleasesStreamOfAbandonedRequest() throws Exception { + conf.set(OzoneClientConfig.OZONE_DOMAIN_SOCKET_PATH, new File(dir, "ozone-socket").getAbsolutePath()); + DomainSocketFactory factory = DomainSocketFactory.getInstance(conf); + XceiverServerDomainSocket server = new XceiverServerDomainSocket(MockDatanodeDetails.randomDatanodeDetails(), + conf, null, readExecutors, metrics, factory); + XceiverClientShortCircuit client = null; + try { + server.setContainerDispatcher(createDispatcherAndPrepareData(new File(dir, "dn-volume"), server, metrics)); + server.start(); + XceiverClientShortCircuit c = new XceiverClientShortCircuit(MockPipeline.createSingleNodePipeline(), conf, + MockDatanodeDetails.randomDatanodeDetails()); + List received = Collections.synchronizedList(new ArrayList<>()); + FieldUtils.writeField(c, "blockStreamCache", new ConcurrentHashMap() { + @Override + public FileInputStream put(Object key, FileInputStream value) { + received.add(value); + return super.put(key, value); + } + }, true); + c.connect(); + // close() fails on a client that never connected, so only a connected client is closed in finally + client = c; + + // the server answers in order on its single read thread, so this delays the responses of the next requests + client.sendCommandInternal(getEchoRequest().toBuilder() + .setClientId(client.getClientId()) + .setCallId(client.getCallId()) + .setEcho(ContainerProtos.EchoRequestProto.newBuilder().setSleepTimeMs(500)) + .build()); + + // the caller stops waiting before the response arrives + ContainerProtos.ContainerCommandRequestProto interrupted = getBlockRequestFor(client); + Thread.currentThread().interrupt(); + assertThatThrownBy(() -> c.sendCommand(interrupted, null)).isInstanceOf(InterruptedIOException.class); + assertThat(Thread.interrupted()).isTrue(); + + // the caller rejects the response after it arrived + ContainerProtos.ContainerCommandRequestProto rejected = getBlockRequestFor(client); + assertThatThrownBy(() -> c.sendCommand(rejected, Collections.singletonList((request, response) -> { + throw new IOException("rejected"); + }))).isInstanceOf(IOException.class).hasMessage("rejected"); + + // responses arrive in order, so the ones above are handled once this one returns + ContainerProtos.ContainerCommandRequestProto claimed = getBlockRequestFor(client); + assertThat(client.sendCommand(claimed, null).getResult()).isEqualTo(ContainerProtos.Result.SUCCESS); + assertThat(received).hasSize(3); + long localID = claimed.getGetBlock().getBlockID().getLocalID(); + try (FileInputStream in = client.getFileInputStream(claimed.getCallId(), localID)) { + assertThat(in).isSameAs(received.get(2)); + assertThat(in.getFD().valid()).isTrue(); + } + assertThat(client.getFileInputStream(interrupted.getCallId(), localID)).isNull(); + assertThat(client.getFileInputStream(rejected.getCallId(), localID)).isNull(); + assertThat(received.get(0).getFD().valid()).isFalse(); + assertThat(received.get(1).getFD().valid()).isFalse(); + } finally { + Thread.interrupted(); + if (client != null) { + client.close(); + } + factory.close(); + server.stop(); + } + } + private static void sendRequest(DataOutputStream outputStream, ContainerProtos.ContainerCommandRequestProto request) throws IOException { outputStream.writeShort(OzoneClientConfig.DATA_TRANSFER_VERSION); @@ -811,6 +889,10 @@ private ContainerProtos.ContainerCommandRequestProto getBlockRequest() { return getBlockRequest(1, true); } + private ContainerProtos.ContainerCommandRequestProto getBlockRequestFor(XceiverClientShortCircuit client) { + return getBlockRequest().toBuilder().setClientId(client.getClientId()).setCallId(client.getCallId()).build(); + } + private ContainerProtos.ContainerCommandRequestProto getBlockRequest(long localID, boolean shortCircuit) { long value = 1; String datanodeUUID = UUID.randomUUID().toString();