This is an automated email from the ASF dual-hosted git repository.
sodonnel pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 1ad6848c3c4 HDDS-16241. gRPC deadline kills long-lived block streams
after 30 seconds and the client never recovers (#11080)
1ad6848c3c4 is described below
commit 1ad6848c3c4b7fd694dddce10f8f36027bf9182e
Author: Sergey Soldatov <[email protected]>
AuthorDate: Mon Oct 5 13:45:11 2026 -0700
HDDS-16241. gRPC deadline kills long-lived block streams after 30 seconds
and the client never recovers (#11080)
---
.../apache/hadoop/hdds/scm/XceiverClientGrpc.java | 5 +-
.../hdds/scm/storage/StreamBlockInputStream.java | 9 ++
.../scm/storage/TestStreamBlockInputStream.java | 46 ++++++
.../common/statemachine/DatanodeConfiguration.java | 22 +++
.../transport/server/GrpcXceiverService.java | 68 ++++++++-
.../common/transport/server/XceiverServerGrpc.java | 5 +-
.../transport/server/TestGrpcXceiverService.java | 169 +++++++++++++++++++++
7 files changed, 320 insertions(+), 4 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
index 4fe56e6ac09..ab9d0dbcafb 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
@@ -618,8 +618,9 @@ public void initStreamRead(BlockID blockID,
StreamingReaderSpi streamObserver,
throw new IOException("Failed to get gRPC stub for DataNode: " + dn);
}
LOG.debug("initStreamRead {} on datanode {}",
blockID.getContainerBlockID(), dn);
- stub.withDeadlineAfter(timeout, TimeUnit.SECONDS)
- .send(new ReadyAwareResponseObserver(dn, streamObserver));
+ // No deadline: it would bound the entire long-lived streaming call.
Per-request timeliness is
+ // enforced by streamReadTimeout in streamRead() and
StreamingReader.poll().
+ stub.send(new ReadyAwareResponseObserver(dn, streamObserver));
return;
} catch (IOException e) {
LOG.error("Failed to start streaming read to DataNode {}", dn, e);
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
index d773701300e..53620567d08 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
@@ -478,6 +478,15 @@ synchronized void readBlockImpl(long length) throws
IOException {
blockID, requestedLength, length, responseDataSize, tokenRef.get(),
pipelineRef.get()), r);
}
+ /**
+ * A stream killed by a gRPC deadline is a transport failure of the
long-lived call, not a data error:
+ * fail over to another datanode the same way as UNAVAILABLE. The classic
per-request path is unaffected.
+ */
+ @Override
+ protected boolean isConnectivityIssue(IOException ex) {
+ return super.isConnectivityIssue(ex) || Status.fromThrowable(ex).getCode()
== Status.DEADLINE_EXCEEDED.getCode();
+ }
+
private void handleExceptions(IOException cause) throws IOException {
IOException root = ConnectionFailureUtils.unwrapCause(cause);
if (root instanceof StorageContainerException || isConnectivityIssue(root)
||
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
index d162b80ad94..ce4a66dea3a 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
@@ -484,6 +484,52 @@ blockID, blockLength, mockStandalonePipeline(), null,
xceiverClientFactory,
}
}
+ /**
+ * A stream killed with DEADLINE_EXCEEDED (e.g. by a deadline on the
long-lived call) must be treated
+ * like any other connectivity failure: fail over to a fresh stream instead
of surfacing the error.
+ */
+ @Test
+ public void testDeadlineExceededFailsOverToFreshStream() throws Exception {
+ OzoneClientConfig clientConfig = newStreamReadConfig();
+ clientConfig.setReadRetryInterval(0);
+ BlockID blockID = new BlockID(1L, 13L);
+ byte[] data = {1, 2, 3, 4};
+ Pipeline pipeline = mockStandalonePipeline();
+ ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver =
mock(ClientCallStreamObserver.class);
+ StreamingReadResponse streamingReadResponse =
mock(StreamingReadResponse.class);
+
when(streamingReadResponse.getRequestObserver()).thenReturn(requestObserver);
+
when(streamingReadResponse.getDatanodeDetails()).thenReturn(MockDatanodeDetails.randomDatanodeDetails());
+
+ AtomicInteger initCount = new AtomicInteger();
+ XceiverClientGrpc xceiverClient = mock(XceiverClientGrpc.class);
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = inv.getArgument(1);
+ reader.setStreamingReadResponse(streamingReadResponse);
+ if (initCount.incrementAndGet() == 1) {
+ // First stream is killed by the transport.
+ reader.onError(Status.DEADLINE_EXCEEDED.asRuntimeException());
+ } else {
+ reader.onNext(ContainerCommandResponseProto.newBuilder()
+ .setCmdType(Type.ReadBlock)
+ .setResult(ContainerProtos.Result.SUCCESS)
+ .setReadBlock(buildReadBlockResponse(data))
+ .build());
+ }
+ return null;
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
+
+ XceiverClientFactory xceiverClientFactory =
mock(XceiverClientFactory.class);
+
when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class))).thenReturn(xceiverClient);
+
+ try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+ blockID, data.length, pipeline, null, xceiverClientFactory,
NO_REFRESH, clientConfig)) {
+ ByteBuffer buf = ByteBuffer.allocate(data.length);
+ assertEquals(data.length, sbis.read(buf), "read should succeed after
failing over to a fresh stream");
+ assertArrayEquals(data, buf.array());
+ }
+ assertEquals(2, initCount.get(), "a fresh stream should have been
initialized for the retry");
+ }
+
private OzoneClientConfig newStreamReadConfig() {
OzoneClientConfig clientConfig = new OzoneClientConfig();
clientConfig.setChecksumVerify(false);
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java
index ea9eb4a1bf5..02861721c31 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java
@@ -152,6 +152,8 @@ public class DatanodeConfiguration extends
ReconfigurableConfig {
public static final String BLOCK_DELETE_COMMAND_WORKER_INTERVAL =
"hdds.datanode.block.delete.command.worker.interval";
public static final Duration BLOCK_DELETE_COMMAND_WORKER_INTERVAL_DEFAULT =
Duration.ofSeconds(2);
+ public static final String STREAM_READ_FILE_IDLE_TIMEOUT_KEY =
"hdds.datanode.stream.read.file.idle.timeout";
+ public static final Duration STREAM_READ_FILE_IDLE_TIMEOUT_DEFAULT =
Duration.ofMinutes(1);
/**
* Number of threads per volume that Datanode will use for chunk read.
@@ -164,6 +166,16 @@ public class DatanodeConfiguration extends
ReconfigurableConfig {
)
private int numReadThreadPerVolume = 10;
+ @Config(key = STREAM_READ_FILE_IDLE_TIMEOUT_KEY,
+ type = ConfigType.TIME,
+ defaultValue = "1m",
+ tags = {DATANODE},
+ description = "How long a streaming ReadBlock stream may stay idle
before the datanode closes the block file "
+ + "held open for it. The stream itself stays open and the file is
reopened by the next request, so this "
+ + "only bounds the file descriptors pinned by idle streams."
+ )
+ private Duration streamReadFileIdleTimeout =
STREAM_READ_FILE_IDLE_TIMEOUT_DEFAULT;
+
/**
* SO_BACKLOG value for the gRPC server socket.
*/
@@ -855,6 +867,12 @@ public void validate() {
BLOCK_DELETE_COMMAND_WORKER_INTERVAL_DEFAULT;
}
+ if (streamReadFileIdleTimeout.isNegative() ||
streamReadFileIdleTimeout.isZero()) {
+ LOG.warn(STREAM_READ_FILE_IDLE_TIMEOUT_KEY + " must be greater than zero
and was set to {}. Defaulting to {}",
+ streamReadFileIdleTimeout, STREAM_READ_FILE_IDLE_TIMEOUT_DEFAULT);
+ streamReadFileIdleTimeout = STREAM_READ_FILE_IDLE_TIMEOUT_DEFAULT;
+ }
+
if (rocksdbLogMaxFileSize < 0) {
LOG.warn(ROCKSDB_LOG_MAX_FILE_SIZE_BYTES_KEY +
" must be no less than zero and was set to {}. Defaulting to {}",
@@ -1211,6 +1229,10 @@ public int getNumReadThreadPerVolume() {
return numReadThreadPerVolume;
}
+ public Duration getStreamReadFileIdleTimeout() {
+ return streamReadFileIdleTimeout;
+ }
+
public void setNumReadThreadPerVolume(int threads) {
this.numReadThreadPerVolume = threads;
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java
index 1516a382688..36e7c8b74f6 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java
@@ -19,7 +19,15 @@
import static
org.apache.hadoop.hdds.protocol.datanode.proto.XceiverClientProtocolServiceGrpc.getSendMethod;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
@@ -47,12 +55,24 @@ public class GrpcXceiverService extends
LOG = LoggerFactory.getLogger(GrpcXceiverService.class);
private final ContainerDispatcher dispatcher;
+ // A streaming ReadBlock call has no deadline, so an idle stream would pin
its block file indefinitely.
+ // This timer closes the file of any stream idle for longer than the
timeout; the next request reopens it.
+ private final long streamReadFileIdleTimeoutNanos;
+ private final ScheduledExecutorService idleFileCloser;
private final ZeroCopyMessageMarshaller<ContainerCommandRequestProto>
zeroCopyMessageMarshaller = new ZeroCopyMessageMarshaller<>(
ContainerCommandRequestProto.getDefaultInstance());
- public GrpcXceiverService(ContainerDispatcher dispatcher) {
+ public GrpcXceiverService(ContainerDispatcher dispatcher, Duration
streamReadFileIdleTimeout,
+ String threadNamePrefix) {
this.dispatcher = dispatcher;
+ this.streamReadFileIdleTimeoutNanos = streamReadFileIdleTimeout.toNanos();
+ this.idleFileCloser = Executors.newSingleThreadScheduledExecutor(new
ThreadFactoryBuilder().setDaemon(true)
+ .setNameFormat(threadNamePrefix + "ReadBlockIdleFileCloser").build());
+ }
+
+ public void shutdown() {
+ idleFileCloser.shutdownNow();
}
/**
@@ -99,15 +119,55 @@ public StreamObserver<ContainerCommandRequestProto> send(
return new StreamObserver<ContainerCommandRequestProto>() {
private final AtomicBoolean isClosed = new AtomicBoolean(false);
private final RandomAccessFileChannel blockFile = new
RandomAccessFileChannel();
+ // Held while a request is served, so the idle closer never closes the
file under an in-flight read.
+ private final ReentrantLock requestLock = new ReentrantLock();
+ private volatile long lastRequestNanos;
+ private volatile ScheduledFuture<?> idleFileCheck;
boolean close() {
if (isClosed.compareAndSet(false, true)) {
+ final ScheduledFuture<?> check = idleFileCheck;
+ if (check != null) {
+ check.cancel(false);
+ }
blockFile.close();
return true;
}
return false;
}
+ private void scheduleIdleFileCheck(long delayNanos) {
+ try {
+ idleFileCheck = idleFileCloser.schedule(this::closeFileIfIdle,
delayNanos, TimeUnit.NANOSECONDS);
+ } catch (RejectedExecutionException e) {
+ // Server is shutting down; the file is closed when the stream ends.
+ LOG.debug("Idle file closer is shut down, not scheduling check", e);
+ }
+ }
+
+ private void closeFileIfIdle() {
+ if (isClosed.get()) {
+ return;
+ }
+ final long idleNanos = System.nanoTime() - lastRequestNanos;
+ if (idleNanos < streamReadFileIdleTimeoutNanos ||
!requestLock.tryLock()) {
+ // Not idle for long enough, or a request is in flight: check again
later.
+ scheduleIdleFileCheck(idleNanos < streamReadFileIdleTimeoutNanos
+ ? streamReadFileIdleTimeoutNanos - idleNanos :
streamReadFileIdleTimeoutNanos);
+ return;
+ }
+ try {
+ if (blockFile.isOpen()) {
+ LOG.debug("Closing block file of streaming read idle for {}ms",
TimeUnit.NANOSECONDS.toMillis(idleNanos));
+ blockFile.close();
+ }
+ // The next request reopens the file and schedules a new check.
+ idleFileCheck = null;
+ } finally {
+ requestLock.unlock();
+ }
+ }
+
@Override
public void onNext(ContainerCommandRequestProto request) {
final DispatcherContext context = request.getCmdType() !=
Type.ReadChunk ? null
@@ -115,6 +175,7 @@ public void onNext(ContainerCommandRequestProto request) {
.setReleaseSupported(true)
.build();
+ requestLock.lock();
try {
if (request.getCmdType() == Type.ReadBlock) {
dispatcher.streamDataReadOnly(request, responseObserver,
blockFile, context);
@@ -132,6 +193,11 @@ public void onNext(ContainerCommandRequestProto request) {
if (context != null) {
context.release();
}
+ lastRequestNanos = System.nanoTime();
+ if (!isClosed.get() && blockFile.isOpen() && idleFileCheck == null) {
+ scheduleIdleFileCheck(streamReadFileIdleTimeoutNanos);
+ }
+ requestLock.unlock();
}
}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java
index 27d697fea75..93078adf2bd 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java
@@ -76,6 +76,7 @@ public final class XceiverServerGrpc implements
XceiverServerSpi {
private DatanodeID id;
private Server server;
private final ContainerDispatcher storageContainer;
+ private final GrpcXceiverService xceiverService;
private boolean isStarted;
private DatanodeDetails datanodeDetails;
private EventLoopGroup eventLoopGroup;
@@ -136,7 +137,8 @@ public XceiverServerGrpc(DatanodeDetails datanodeDetails,
}
LOG.info("GrpcServer channel type {}", channelType.getSimpleName());
- GrpcXceiverService xceiverService = new GrpcXceiverService(dispatcher);
+ xceiverService = new GrpcXceiverService(dispatcher,
dnConf.getStreamReadFileIdleTimeout(),
+ datanodeDetails.threadNamePrefix());
NettyServerBuilder nettyServerBuilder = NettyServerBuilder.forPort(port)
.maxInboundMessageSize(OzoneConsts.OZONE_SCM_CHUNK_MAX_SIZE)
.bossEventLoopGroup(eventLoopGroup)
@@ -234,6 +236,7 @@ public void stop() {
try {
server.shutdown();
server.awaitTermination(5, TimeUnit.SECONDS);
+ xceiverService.shutdown();
eventLoopGroup.shutdownGracefully().sync();
} catch (InterruptedException e) {
LOG.error("failed to shutdown XceiverServerGrpc", e);
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/TestGrpcXceiverService.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/TestGrpcXceiverService.java
new file mode 100644
index 00000000000..1d6fad6f533
--- /dev/null
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/TestGrpcXceiverService.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.common.transport.server;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicReference;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadBlockRequestProto;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
+import org.apache.hadoop.hdds.utils.io.RandomAccessFileChannel;
+import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
+import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Tests for the streaming ReadBlock handling in {@link GrpcXceiverService}:
the block file held open for a
+ * stream is closed once the stream is idle and reopened by the next request.
+ */
+class TestGrpcXceiverService {
+
+ private static final Duration IDLE_TIMEOUT = Duration.ofMillis(200);
+
+ @TempDir
+ private Path tempDir;
+
+ private GrpcXceiverService service;
+
+ @AfterEach
+ void shutdown() {
+ if (service != null) {
+ service.shutdown();
+ }
+ }
+
+ private static ContainerCommandRequestProto readBlockRequest() {
+ return ContainerCommandRequestProto.newBuilder()
+ .setCmdType(Type.ReadBlock)
+ .setContainerID(1)
+ .setDatanodeUuid("dn")
+ .setReadBlock(ReadBlockRequestProto.newBuilder()
+
.setBlockID(DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1))
+ .setOffset(0))
+ .build();
+ }
+
+ /**
+ * Mimics {@code KeyValueHandler.readBlockImpl}: open the block file on
first use, then serve the request.
+ * The optional hook runs while the request is in flight.
+ */
+ private ContainerDispatcher mockDispatcher(File blockFile,
AtomicReference<RandomAccessFileChannel> channelRef,
+ Runnable inFlight) throws Exception {
+ ContainerDispatcher dispatcher = mock(ContainerDispatcher.class);
+ doAnswer(inv -> {
+ RandomAccessFileChannel channel = inv.getArgument(2);
+ channelRef.set(channel);
+ if (!channel.isOpen()) {
+ channel.open(blockFile);
+ }
+ inFlight.run();
+ return null;
+ }).when(dispatcher).streamDataReadOnly(any(), any(), any(), any());
+ return dispatcher;
+ }
+
+ @Test
+ void idleStreamClosesBlockFileAndNextRequestReopensIt() throws Exception {
+ File blockFile = Files.createFile(tempDir.resolve("block")).toFile();
+ AtomicReference<RandomAccessFileChannel> channelRef = new
AtomicReference<>();
+ // Checked while the request is in flight: after onNext returns, the idle
closer may already have run.
+ AtomicReference<Boolean> openInFlight = new AtomicReference<>();
+ Runnable recordOpen = () -> openInFlight.set(channelRef.get().isOpen());
+ service = new GrpcXceiverService(mockDispatcher(blockFile, channelRef,
recordOpen), IDLE_TIMEOUT, "test-");
+ StreamObserver<ContainerCommandResponseProto> responseObserver =
mock(StreamObserver.class);
+ StreamObserver<ContainerCommandRequestProto> requestObserver =
service.send(responseObserver);
+
+ requestObserver.onNext(readBlockRequest());
+ RandomAccessFileChannel channel = channelRef.get();
+ assertNotNull(channel);
+ assertTrue(openInFlight.get(), "block file should be open while a request
is served");
+
+ GenericTestUtils.waitFor(() -> !channel.isOpen(), 20, 5000);
+ verify(responseObserver, never()).onError(any());
+ verify(responseObserver, never()).onCompleted();
+
+ openInFlight.set(null);
+ requestObserver.onNext(readBlockRequest());
+ assertTrue(openInFlight.get(), "next request should reopen the block
file");
+ GenericTestUtils.waitFor(() -> !channel.isOpen(), 20, 5000);
+
+ requestObserver.onCompleted();
+ assertFalse(channel.isOpen());
+ verify(responseObserver).onCompleted();
+ }
+
+ @Test
+ void inFlightRequestKeepsBlockFileOpen() throws Exception {
+ File blockFile = Files.createFile(tempDir.resolve("block")).toFile();
+ AtomicReference<RandomAccessFileChannel> channelRef = new
AtomicReference<>();
+ AtomicReference<Boolean> openDuringSlowRequest = new AtomicReference<>();
+ Runnable slowRequest = () -> {
+ // Stay in flight well past the idle timeout, then record whether the
file survived.
+ try {
+ Thread.sleep(3 * IDLE_TIMEOUT.toMillis());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ openDuringSlowRequest.set(channelRef.get().isOpen());
+ };
+ AtomicReference<Runnable> inFlight = new AtomicReference<>(() -> { });
+ service = new GrpcXceiverService(mockDispatcher(blockFile, channelRef, ()
-> inFlight.get().run()),
+ IDLE_TIMEOUT, "test-");
+ StreamObserver<ContainerCommandRequestProto> requestObserver =
service.send(mock(StreamObserver.class));
+
+ // First request arms the idle check; the second one is still running when
it fires.
+ requestObserver.onNext(readBlockRequest());
+ inFlight.set(slowRequest);
+ requestObserver.onNext(readBlockRequest());
+ assertTrue(openDuringSlowRequest.get(), "an in-flight request must not
lose its block file");
+
+ RandomAccessFileChannel channel = channelRef.get();
+ GenericTestUtils.waitFor(() -> !channel.isOpen(), 20, 5000);
+ requestObserver.onCompleted();
+ }
+
+ @Test
+ void errorClosesBlockFileImmediately() throws Exception {
+ File blockFile = Files.createFile(tempDir.resolve("block")).toFile();
+ AtomicReference<RandomAccessFileChannel> channelRef = new
AtomicReference<>();
+ service = new GrpcXceiverService(mockDispatcher(blockFile, channelRef, ()
-> { }), Duration.ofHours(1), "test-");
+ StreamObserver<ContainerCommandRequestProto> requestObserver =
service.send(mock(StreamObserver.class));
+
+ requestObserver.onNext(readBlockRequest());
+ assertTrue(channelRef.get().isOpen());
+ requestObserver.onError(new RuntimeException("client went away"));
+ assertFalse(channelRef.get().isOpen(), "stream teardown must close the
block file");
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
