This is an automated email from the ASF dual-hosted git repository.
sadanand48 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 2076d7facfc HDDS-15521. StreamBlockInputStream fails with
TimeoutIOException without retry or failover. (#10479)
2076d7facfc is described below
commit 2076d7facfcd2933969b58c779c44d2f505d6fc9
Author: Sadanand Shenoy <[email protected]>
AuthorDate: Tue Jul 7 14:13:33 2026 +0530
HDDS-15521. StreamBlockInputStream fails with TimeoutIOException without
retry or failover. (#10479)
---
.../apache/hadoop/hdds/scm/XceiverClientGrpc.java | 14 +
.../hdds/scm/storage/StreamBlockInputStream.java | 51 +++-
.../scm/storage/TestStreamBlockInputStream.java | 10 +-
.../hadoop/hdds/utils/ConnectionFailureUtils.java | 28 ++
.../rpc/read/TestStreamReadDatanodeFailover.java | 287 +++++++++++++++++++++
5 files changed, 375 insertions(+), 15 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 90e581e8f4b..8c52f339027 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
@@ -30,6 +30,7 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -603,9 +604,22 @@ public void streamRead(ContainerCommandRequestProto
request,
@Override
public void initStreamRead(BlockID blockID, StreamingReaderSpi
streamObserver) throws IOException {
+ initStreamRead(blockID, streamObserver, Collections.emptySet());
+ }
+
+ /**
+ * Start a streaming read, skipping datanodes that previously failed for
this block stream.
+ */
+ public void initStreamRead(BlockID blockID, StreamingReaderSpi
streamObserver,
+ Set<DatanodeID> excludedDatanodes) throws IOException {
final List<DatanodeDetails> datanodeList = sortDatanodes(null,
ContainerProtos.Type.ReadBlock);
IOException lastException = null;
for (DatanodeDetails dn : datanodeList) {
+ if (excludedDatanodes.contains(dn.getID())) {
+ LOG.debug("Skipping excluded datanode {} (uuid={}) for initStreamRead
{}",
+ dn, dn.getUuidString(), blockID.getContainerBlockID());
+ continue;
+ }
try {
checkOpen(dn);
semaphore.acquire();
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 dd3928f9137..975c0511832 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
@@ -23,6 +23,8 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.Duration;
+import java.util.HashSet;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -36,6 +38,8 @@
import org.apache.hadoop.fs.FSExceptionMessages;
import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.DatanodeID;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadBlockResponseProto;
import org.apache.hadoop.hdds.scm.OzoneClientConfig;
@@ -47,6 +51,7 @@
import
org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
+import org.apache.hadoop.hdds.utils.ConnectionFailureUtils;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.ozone.common.Checksum;
import org.apache.hadoop.ozone.common.ChecksumData;
@@ -91,6 +96,7 @@ public class StreamBlockInputStream extends
BlockExtendedInputStream {
private final Function<BlockID, BlockLocationInfo> refreshFunction;
private final RetryPolicy retryPolicy;
private int retries = 0;
+ private final Set<DatanodeID> failedStreamingDatanodes = new HashSet<>();
public StreamBlockInputStream(
BlockID blockID, long length, Pipeline pipeline,
@@ -171,13 +177,19 @@ private synchronized boolean dataAvailableToRead(int
length, boolean preRead) th
if (position >= blockLength) {
return false;
}
- initialize();
-
- if (bufferHasRemaining()) {
- return true;
+ while (true) {
+ try {
+ initialize();
+ if (bufferHasRemaining()) {
+ return true;
+ }
+ buffer = streamingReader.read(length, preRead);
+ retries = 0;
+ return bufferHasRemaining();
+ } catch (IOException ex) {
+ handleExceptions(ex);
+ }
}
- buffer = streamingReader.read(length, preRead);
- return bufferHasRemaining();
}
private synchronized void advancePosition(long delta) {
@@ -293,7 +305,7 @@ private synchronized void initialize() throws IOException {
try {
acquireClient();
final StreamingReader reader = new StreamingReader();
- xceiverClient.initStreamRead(blockID, reader);
+ xceiverClient.initStreamRead(blockID, reader,
failedStreamingDatanodes);
streamingReader = reader;
} catch (IOException ioe) {
handleExceptions(ioe);
@@ -327,10 +339,14 @@ synchronized void readBlockImpl(long length) throws
IOException {
}
private void handleExceptions(IOException cause) throws IOException {
- if (cause instanceof StorageContainerException ||
isConnectivityIssue(cause)) {
- if (shouldRetryRead(cause, retryPolicy, retries++)) {
+ IOException root = ConnectionFailureUtils.unwrapCause(cause);
+ if (root instanceof StorageContainerException || isConnectivityIssue(root)
||
+ root instanceof TimeoutIOException) {
+ if (shouldRetryRead(root, retryPolicy, retries++)) {
+ recordFailedStreamingDatanode();
releaseClient();
- refreshBlockInfo(cause);
+ refreshBlockInfo(root);
+ requestedLength = position;
LOG.warn("Refreshing block data to read block {} due to {}", blockID,
cause.getMessage());
} else {
throw cause;
@@ -340,6 +356,21 @@ private void handleExceptions(IOException cause) throws
IOException {
}
}
+ private void recordFailedStreamingDatanode() {
+ if (streamingReader == null) {
+ return;
+ }
+ final StreamingReadResponse response = streamingReader.getResponse();
+ if (response == null) {
+ return;
+ }
+ final DatanodeDetails dn = response.getDatanodeDetails();
+ if (failedStreamingDatanodes.add(dn.getID())) {
+ LOG.warn("Excluding DataNode {} from streaming read retries for block
{}",
+ dn, blockID);
+ }
+ }
+
protected synchronized void releaseClient() {
if (xceiverClientFactory != null && xceiverClient != null) {
closeStream();
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 4cdf13032b4..9e1ff0893ec 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
@@ -224,7 +224,7 @@ public void
testPollDoesNotDropQueuedItemWhenFutureCompletesFirst() throws Excep
reader.setStreamingReadResponse(streamingReadResponse);
readerRef.set(reader);
return null;
- }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
// Simulate the race: when the client sends a ReadBlock request, the server
// responds with data (onNext) and closes the stream (onCompleted) before
@@ -304,7 +304,7 @@ private XceiverClientGrpc mockStreamingReadClient(byte[]
data,
.setReadBlock(readBlock)
.build());
return null;
- }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
return xceiverClient;
}
@@ -338,7 +338,7 @@ public void
testReadDoesNotDropQueuedItemsWhenFutureIsDoneOnSecondCall() throws
reader.setStreamingReadResponse(streamingReadResponse);
readerRef.set(reader);
return null;
- }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
// Server delivers both 4-byte chunks plus onCompleted() in one synchronous
// call. After streamRead() returns: queue=[chunk1, chunk2], isDone=true.
@@ -388,7 +388,7 @@ public void
testReadGetsFreshResponseTimeoutAfterStreamReadWait() throws Excepti
reader.setStreamingReadResponse(streamingReadResponse);
readerRef.set(reader);
return null;
- }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
doAnswer(inv -> {
Thread.sleep(450);
Thread responseThread = new Thread(() -> {
@@ -437,7 +437,7 @@ public void
testReadWithoutNewRequestGetsFreshTimeoutBudget() throws Exception {
reader.setStreamingReadResponse(streamingReadResponse);
readerRef.set(reader);
return null;
- }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any(), any());
doAnswer(inv -> {
streamReads.incrementAndGet();
readerRef.get().onNext(buildResponseProto(new byte[] {1}, 0));
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
index 62cd868c763..d7f45b1cd7e 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
@@ -18,11 +18,15 @@
package org.apache.hadoop.hdds.utils;
import java.io.EOFException;
+import java.io.IOException;
import java.net.ConnectException;
import java.net.NoRouteToHostException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
+import java.util.concurrent.ExecutionException;
+import
org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
+import org.apache.ratis.protocol.exceptions.TimeoutIOException;
/**
* Shared classifier for exceptions where the cached peer IP is no longer
@@ -100,4 +104,28 @@ public static boolean isConnectionFailure(Throwable t) {
}
return false;
}
+
+ /**
+ * Returns the first {@link StorageContainerException} or
+ * {@link TimeoutIOException} in {@code ex}'s cause chain (through
+ * {@link ExecutionException} and nested {@link IOException} wrappers).
+ */
+ public static IOException unwrapCause(IOException ex) {
+ Throwable t = ex;
+ while (t != null) {
+ if (t instanceof TimeoutIOException || t instanceof
StorageContainerException) {
+ return (IOException) t;
+ }
+ if (t instanceof ExecutionException && t.getCause() != null) {
+ t = t.getCause();
+ continue;
+ }
+ if (t.getCause() instanceof IOException) {
+ t = t.getCause();
+ continue;
+ }
+ break;
+ }
+ return ex;
+ }
}
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamReadDatanodeFailover.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamReadDatanodeFailover.java
new file mode 100644
index 00000000000..27a971873be
--- /dev/null
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamReadDatanodeFailover.java
@@ -0,0 +1,287 @@
+/*
+ * 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.client.rpc.read;
+
+import static
org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE;
+import static
org.apache.hadoop.ozone.client.OzoneClientTestUtils.assertKeyContent;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.StreamingReadResponse;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.scm.storage.StreamBlockInputStream;
+import org.apache.hadoop.hdds.utils.IOUtils;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.TestDataUtil;
+import org.apache.hadoop.ozone.client.ObjectStore;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.OzoneClientFactory;
+import org.apache.hadoop.ozone.client.io.KeyInputStream;
+import org.apache.hadoop.ozone.om.TestBucket;
+import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+/**
+ * Verifies that streaming block reads fail over to a healthy replica when the
+ * datanode serving the stream becomes unavailable, matching legacy
+ * {@code BlockInputStream} behavior.
+ *
+ * <p>With {@code ozone.client.stream.readblock.enable=true}, reads currently
+ * do not fail over correctly when the streaming datanode stops responding.
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class TestStreamReadDatanodeFailover {
+
+ private static final Duration STREAM_READ_TIMEOUT = Duration.ofSeconds(3);
+ private static final int PIPELINE_READY_TIMEOUT_MS = 30_000;
+
+ private MiniOzoneCluster cluster;
+ private final Set<DatanodeDetails> stoppedDatanodes = new HashSet<>();
+
+ @BeforeAll
+ void setup() throws Exception {
+ cluster = newCluster();
+ cluster.waitForClusterToBeReady();
+ }
+
+ @BeforeEach
+ void waitForPipeline() throws Exception {
+ cluster.waitForPipelineTobeReady(THREE, PIPELINE_READY_TIMEOUT_MS);
+ }
+
+ @AfterEach
+ void resetDatanodes() throws Exception {
+ if (stoppedDatanodes.isEmpty()) {
+ return;
+ }
+ List<DatanodeDetails> toRestart = new ArrayList<>(stoppedDatanodes);
+ stoppedDatanodes.clear();
+ for (DatanodeDetails dn : toRestart) {
+ cluster.restartHddsDatanode(dn, false);
+ }
+ cluster.waitForClusterToBeReady();
+ }
+
+ @AfterAll
+ void cleanup() {
+ IOUtils.closeQuietly(cluster);
+ }
+
+ /**
+ * Legacy (Async) chunk reads succeed after stopping one pipeline datanode.
+ */
+ @Test
+ void testAsyncReadFailoverWhenDatanodeStopped() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ ObjectStore store = client.getObjectStore();
+ OzoneBucket bucket = createBucket(store);
+ String keyName = newKeyName();
+ byte[] content = RandomUtils.secure().randomBytes(32 * 1024);
+ TestDataUtil.createKey(bucket, keyName,
+ RatisReplicationConfig.getInstance(THREE), content);
+
+ List<DatanodeDetails> datanodes = getPipelineDatanodes(cluster, bucket,
keyName);
+ assertEquals(3, datanodes.size());
+
+ stopDatanode(datanodes.get(0));
+ assertKeyContent(bucket, keyName, content);
+
+ stopDatanode(datanodes.get(1));
+ assertKeyContent(bucket, keyName, content);
+ }
+ }
+
+ /**
+ * Streaming reads should succeed after stopping pipeline datanodes when at
+ * least one healthy replica remains, same as the legacy path.
+ */
+ @Test
+ void testStreamReadFailoverWhenDatanodesStopped() throws Exception {
+ OzoneConfiguration streamConf = streamReadConfig(cluster.getConf());
+ try (OzoneClient client = OzoneClientFactory.getRpcClient(streamConf)) {
+ ObjectStore store = client.getObjectStore();
+ OzoneBucket bucket = createBucket(store);
+ String keyName = newKeyName();
+ byte[] content = RandomUtils.secure().randomBytes(32 * 1024);
+ TestDataUtil.createKey(bucket, keyName,
+ RatisReplicationConfig.getInstance(THREE), content);
+
+ List<DatanodeDetails> datanodes = getPipelineDatanodes(cluster, bucket,
keyName);
+ assertEquals(3, datanodes.size());
+
+ // Sanity check: streaming read works with all datanodes up.
+ assertKeyContent(bucket, keyName, content);
+
+ stopDatanode(datanodes.get(0));
+ assertKeyContent(bucket, keyName, content);
+
+ stopDatanode(datanodes.get(1));
+ assertKeyContent(bucket, keyName, content);
+ }
+ }
+
+ /**
+ * After a streaming read has started, stopping the datanode serving the
+ * stream must not break the read. Legacy reads fail over and continue from
+ * the current offset; streaming reads currently fail (timeout or wrong
data).
+ */
+ @Test
+ void testStreamReadFailoverAfterActiveDatanodeStopped() throws Exception {
+ OzoneConfiguration streamConf = streamReadConfig(cluster.getConf());
+ try (OzoneClient streamClient =
OzoneClientFactory.getRpcClient(streamConf);
+ OzoneClient legacyClient = cluster.newClient()) {
+ TestBucket streamBucket = TestBucket.newBuilder(streamClient).build();
+ OzoneBucket legacyBucket = legacyClient.getObjectStore()
+ .getVolume(streamBucket.delegate().getVolumeName())
+ .getBucket(streamBucket.delegate().getName());
+
+ String keyName = newKeyName();
+ byte[] content = streamBucket.writeRandomBytes(keyName, 32 * 1024);
+
+ List<DatanodeDetails> datanodes =
+ getPipelineDatanodes(cluster, streamBucket.delegate(), keyName);
+ assertEquals(3, datanodes.size());
+
+ // Legacy control: stopping one pipeline datanode mid-read still
succeeds.
+ try (InputStream legacyIn = legacyBucket.readKey(keyName)) {
+ assertNotEquals(-1, legacyIn.read());
+ stopDatanode(datanodes.get(0));
+ readRemaining(legacyIn, content, 1);
+ }
+
+ // Streaming read: stop the datanode actively serving the stream.
+ // This fails today without a proper streaming read failover fix.
+ try (KeyInputStream streamIn = streamBucket.getKeyInputStream(keyName)) {
+ assertNotEquals(-1, streamIn.read());
+ StreamBlockInputStream blockStream =
+ (StreamBlockInputStream) streamIn.getPartStreams().get(0);
+ DatanodeDetails activeDatanode =
getActiveStreamingDatanode(blockStream);
+ assertTrue(datanodes.contains(activeDatanode),
+ "Active streaming datanode should belong to the key pipeline");
+ stopDatanode(activeDatanode);
+ readRemaining(streamIn, content, 1);
+ }
+ }
+ }
+
+ private void stopDatanode(DatanodeDetails dn) throws IOException {
+ cluster.shutdownHddsDatanode(dn);
+ stoppedDatanodes.add(dn);
+ }
+
+ private static void readRemaining(InputStream in, byte[] expected, int
offset)
+ throws IOException {
+ byte[] actual = org.apache.commons.io.IOUtils.readFully(in,
expected.length - offset);
+ assertArrayEquals(
+ java.util.Arrays.copyOfRange(expected, offset, expected.length),
+ actual);
+ }
+
+ private static DatanodeDetails
getActiveStreamingDatanode(StreamBlockInputStream blockStream)
+ throws ReflectiveOperationException {
+ Field streamingReaderField =
StreamBlockInputStream.class.getDeclaredField("streamingReader");
+ streamingReaderField.setAccessible(true);
+ Object streamingReader = streamingReaderField.get(blockStream);
+ assertTrue(streamingReader != null, "Streaming reader should be
initialized after first read");
+
+ Method getResponse =
streamingReader.getClass().getDeclaredMethod("getResponse");
+ getResponse.setAccessible(true);
+ StreamingReadResponse response = (StreamingReadResponse)
getResponse.invoke(streamingReader);
+ assertTrue(response != null, "Streaming read response should be
initialized");
+ return response.getDatanodeDetails();
+ }
+
+ private static OzoneConfiguration streamReadConfig(OzoneConfiguration base) {
+ OzoneClientConfig clientConfig = base.getObject(OzoneClientConfig.class);
+ clientConfig.setStreamReadBlock(true);
+ clientConfig.setStreamReadTimeout(STREAM_READ_TIMEOUT);
+ OzoneConfiguration conf = new OzoneConfiguration(base);
+ conf.setFromObject(clientConfig);
+ return conf;
+ }
+
+ private static MiniOzoneCluster newCluster() throws IOException {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.setInt(ScmConfigKeys.OZONE_SCM_PIPELINE_OWNER_CONTAINER_COUNT, 1);
+ conf.setInt(ScmConfigKeys.OZONE_DATANODE_PIPELINE_LIMIT, 1);
+ conf.setInt(ScmConfigKeys.OZONE_SCM_RATIS_PIPELINE_LIMIT, 5);
+ return MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(3)
+ .build();
+ }
+
+ private static OzoneBucket createBucket(ObjectStore store) throws
IOException {
+ String volumeName = UUID.randomUUID().toString();
+ store.createVolume(volumeName);
+ String bucketName = UUID.randomUUID().toString();
+ store.getVolume(volumeName).createBucket(bucketName);
+ return store.getVolume(volumeName).getBucket(bucketName);
+ }
+
+ private static String newKeyName() {
+ return "key-" + UUID.randomUUID();
+ }
+
+ private static List<DatanodeDetails> getPipelineDatanodes(MiniOzoneCluster
cluster,
+ OzoneBucket bucket, String keyName) throws IOException {
+ OmKeyArgs keyArgs = new OmKeyArgs.Builder()
+ .setVolumeName(bucket.getVolumeName())
+ .setBucketName(bucket.getName())
+ .setKeyName(keyName)
+ .build();
+ OmKeyLocationInfo keyInfo = cluster.getOzoneManager().lookupKey(keyArgs)
+ .getKeyLocationVersions().get(0)
+ .getBlocksLatestVersionOnly().get(0);
+ long containerID = keyInfo.getContainerID();
+
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ ContainerInfo container = scm.getContainerManager()
+ .getContainer(ContainerID.valueOf(containerID));
+ Pipeline pipeline = scm.getPipelineManager()
+ .getPipeline(container.getPipelineID());
+ return pipeline.getNodes();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
