diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ContentGenerator.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ContentGenerator.java index 542634c4884b..8adf0cde4e5a 100644 --- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ContentGenerator.java +++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/ContentGenerator.java @@ -18,10 +18,12 @@ import java.io.IOException; import java.io.OutputStream; +import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import com.google.common.annotations.VisibleForTesting; import org.apache.commons.lang3.RandomStringUtils; +import org.apache.hadoop.ozone.client.io.OzoneDataStreamOutput; /** * Utility class to write random keys from a limited buffer. @@ -80,6 +82,22 @@ public void write(OutputStream outputStream) throws IOException { } } + /** + * Write the required bytes to the streaming output stream. + */ + public void write(OzoneDataStreamOutput out) throws IOException { + for (long nrRemaining = keySize; + nrRemaining > 0; nrRemaining -= bufferSize) { + int curSize = (int) Math.min(bufferSize, nrRemaining); + for (int i = 0; i < curSize; i += copyBufferSize) { + ByteBuffer bb = + ByteBuffer.wrap(buffer, i, Math.min(copyBufferSize, curSize - i)); + out.write(bb); + } + } + out.close(); + } + @VisibleForTesting byte[] getBuffer() { return buffer; diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/OzoneClientKeyGenerator.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/OzoneClientKeyGenerator.java index cf5e2604e783..d249d456ef01 100644 --- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/OzoneClientKeyGenerator.java +++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/OzoneClientKeyGenerator.java @@ -22,13 +22,16 @@ import java.util.concurrent.Callable; import org.apache.hadoop.hdds.cli.HddsVersionProvider; +import org.apache.hadoop.hdds.client.ReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationFactor; import org.apache.hadoop.hdds.client.ReplicationType; import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.ozone.client.OzoneBucket; import org.apache.hadoop.ozone.client.OzoneClient; import com.codahale.metrics.Timer; +import org.apache.hadoop.ozone.client.io.OzoneDataStreamOutput; import picocli.CommandLine.Command; import picocli.CommandLine.Option; @@ -78,6 +81,12 @@ public class OzoneClientKeyGenerator extends BaseFreonGenerator ) private String omServiceID = null; + @Option( + names = {"--enable-streaming", "--stream"}, + description = "Specify whether the write will be through ratis streaming" + ) + private boolean enableRatisStreaming = false; + private Timer timer; private OzoneBucket bucket; @@ -102,7 +111,11 @@ public Void call() throws Exception { timer = getMetrics().timer("key-create"); - runTests(this::createKey); + if (enableRatisStreaming) { + runTests(this::createStreamKey); + } else { + runTests(this::createKey); + } } return null; } @@ -119,4 +132,19 @@ private void createKey(long counter) throws Exception { return null; }); } + + private void createStreamKey(long counter) throws Exception { + final ReplicationConfig replicationConfig = ReplicationConfig + .fromProtoTypeAndFactor(HddsProtos.ReplicationType.RATIS, + HddsProtos.ReplicationFactor.THREE); + final String key = generateObjectName(counter); + + timer.time(() -> { + try (OzoneDataStreamOutput stream = bucket + .createStreamKey(key, keySize, replicationConfig, metadata)) { + contentGenerator.write(stream); + } + return null; + }); + } }