feat(java): add actor and data batch generator for java bench

foreign/java/bench/src/main/java/org/apache/iggy/bench/benchmarks/actors/tcp/async/TcpAsyncPinnedProducerActor.java

@@ -0,0 +1,121 @@
+/*
+ * 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.iggy.bench.benchmarks.actors.tcp.async;
+
+import org.apache.iggy.bench.models.cli.GlobalCliArgs;
+import org.apache.iggy.bench.models.common.generator.DataBatch;
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Partitioning;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
+
+public final class TcpAsyncPinnedProducerActor {
+
+    private static final long PARTITION_ID = 0L;
+
+    private final GlobalCliArgs globalCliArgs;
+    private final int actorId;
+    private final StreamId streamId;
+    private final TopicId topicId;
+    private final Partitioning partitioning;
+    private final DataBatch fullBatch;
+    private final long targetMessageBatches;
+    private final long targetDataBytes;
+    private AsyncIggyTcpClient client;
+
+    public TcpAsyncPinnedProducerActor(
+            GlobalCliArgs globalCliArgs,
+            int actorId,
+            String streamName,
+            String topicName,
+            DataBatch fullBatch,
+            long targetMessageBatches,
+            long targetDataBytes) {
+        this.globalCliArgs = globalCliArgs;
+        this.actorId = actorId;
+        this.streamId = StreamId.of(streamName);
+        this.topicId = TopicId.of(topicName);
+        this.partitioning = Partitioning.partitionId(PARTITION_ID);
+        this.fullBatch = fullBatch;
+        this.targetMessageBatches = targetMessageBatches;
+        this.targetDataBytes = targetDataBytes;
+    }
+
+    public CompletableFuture<Void> run() {
+        try {
+            return AsyncIggyTcpClient.builder()
+                    .credentials(globalCliArgs.username(), globalCliArgs.password())
+                    .buildAndLogin()
+                    .thenCompose(client -> {
+                        this.client = client;
+                        long warmupDeadline = globalCliArgs.warmupTimeMs() <= 0L
+                                ? 0L
+                                : System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(globalCliArgs.warmupTimeMs());
+
+                        return sendMessages(warmupDeadline, 0L, 0L)
+                                .handle((ignored, sendFailure) -> {
+                                    CompletableFuture<Void> closeFuture = client.close();
+
+                                    if (sendFailure != null) {
+                                        CompletableFuture<Void> failedAfterClose =
+                                                closeFuture.handle((closeIgnored, closeFailure) -> {
+                                                    throw new CompletionException(sendFailure);
+                                                });
+                                        return failedAfterClose;
+                                    }
+
+                                    return closeFuture;
+                                })
+                                .thenCompose(future -> future);
+                    });
+        } catch (RuntimeException exception) {
+            return CompletableFuture.failedFuture(exception);
+        }
+    }
+
+    private CompletableFuture<Void> sendMessages(long warmupDeadline, long sentBatches, long sentBytes) {
+        if (warmupDeadline > 0L) {
+            if (System.nanoTime() >= warmupDeadline) {
+                return sendMessages(0L, 0L, 0L);
+            }
+
+            return sendBatch(fullBatch, false)
+                    .thenCompose(ignored -> sendMessages(warmupDeadline, sentBatches, sentBytes));
+        }
+
+        if (targetDataBytes > 0L ? sentBytes >= targetDataBytes : sentBatches >= targetMessageBatches) {
+            return CompletableFuture.completedFuture(null);
+        }
+
+        return sendBatch(fullBatch, true).thenCompose(ignored -> {
+            long updatedSentBatches = sentBatches + 1L;
+            long updatedSentBytes = sentBytes + fullBatch.userDataBytes();
+            return sendMessages(0L, updatedSentBatches, updatedSentBytes);
+        });
+    }
+
+    private CompletableFuture<Void> sendBatch(DataBatch currentBatch, boolean recordMetrics) {
+        return client.messages().sendMessages(streamId, topicId, partitioning, currentBatch.messages());
+    }
+}

foreign/java/bench/src/main/java/org/apache/iggy/bench/benchmarks/runners/tcp/async/TcpAsyncPinnedProducer.java

@@ -0,0 +1,88 @@
+/*
+ * 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.iggy.bench.benchmarks.runners.tcp.async;
+
+import org.apache.iggy.bench.benchmarks.actors.tcp.async.TcpAsyncPinnedProducerActor;
+import org.apache.iggy.bench.common.generator.BenchmarkBatchGenerator;
+import org.apache.iggy.bench.common.provision.ResourceProvisioner;
+import org.apache.iggy.bench.models.cli.GlobalCliArgs;
+import org.apache.iggy.bench.models.cli.PinnedProducerCliArgs;
+import org.apache.iggy.bench.models.common.generator.DataBatch;
+import org.apache.iggy.bench.models.common.provision.ProvisionedResources;
+
+import java.util.concurrent.CompletableFuture;
+
+public final class TcpAsyncPinnedProducer {
+
+    private final GlobalCliArgs globalCliArgs;
+    private final PinnedProducerCliArgs pinnedProducerCliArgs;
+    private final ResourceProvisioner resourceProvisioner;
+    private ProvisionedResources provisionedResources;
+
+    TcpAsyncPinnedProducer(
+            GlobalCliArgs globalCliArgs,
+            PinnedProducerCliArgs pinnedProducerCliArgs,
+            ResourceProvisioner resourceProvisioner) {
+        this.globalCliArgs = globalCliArgs;
+        this.pinnedProducerCliArgs = pinnedProducerCliArgs;
+        this.resourceProvisioner = resourceProvisioner;
+
+        provisionResources();
+        runBenchmark().join();
+    }
+
+    public TcpAsyncPinnedProducer(GlobalCliArgs globalCliArgs, PinnedProducerCliArgs pinnedProducerCliArgs) {
+        this(globalCliArgs, pinnedProducerCliArgs, new ResourceProvisioner());
+    }
+
+    private void provisionResources() {
+        this.provisionedResources = resourceProvisioner.provisionResources(globalCliArgs, pinnedProducerCliArgs);
+    }
+
+    private CompletableFuture<Void> runBenchmark() {
+        try {
+            String topicName = provisionedResources.topicNames().get(0);
+            var batchGenerator =
+                    new BenchmarkBatchGenerator(globalCliArgs.messageSize(), globalCliArgs.messagesPerBatch());
+            DataBatch fullBatch = batchGenerator.generateBatch();
+            long targetMessageBatches = globalCliArgs.totalData() > 0L ? 0L : globalCliArgs.messageBatches();
+            long targetDataBytes =
+                    globalCliArgs.totalData() > 0L ? globalCliArgs.totalData() / pinnedProducerCliArgs.producers() : 0L;
+            var actorRuns = new CompletableFuture<?>[pinnedProducerCliArgs.producers()];
+
+            for (int index = 0; index < pinnedProducerCliArgs.producers(); index++) {
+                String streamName = provisionedResources.streamNames().get(index);
+                var actor = new TcpAsyncPinnedProducerActor(
+                        globalCliArgs,
+                        index + 1,
+                        streamName,
+                        topicName,
+                        fullBatch,
+                        targetMessageBatches,
+                        targetDataBytes);
+                actorRuns[index] = actor.run();
+            }
+
+            return CompletableFuture.allOf(actorRuns);
+        } catch (RuntimeException exception) {
+            return CompletableFuture.failedFuture(exception);
+        }
+    }
+}

foreign/java/bench/src/main/java/org/apache/iggy/bench/cli/PinnedProducerCommand.java

@@ -19,7 +19,7 @@
 
 package org.apache.iggy.bench.cli;
 
-import org.apache.iggy.bench.benchmarks.tcp.async.TcpAsyncPinnedProducer;
+import org.apache.iggy.bench.benchmarks.runners.tcp.async.TcpAsyncPinnedProducer;
 import org.apache.iggy.bench.models.cli.GlobalCliArgs;
 import org.apache.iggy.bench.models.cli.PinnedProducerCliArgs;
 import org.slf4j.Logger;
@@ -97,9 +97,7 @@ public Integer call() {
             pinnedProducerCliArgs.validate();
 
             log.info("Starting the Pinned Producer benchmark...");
-            var benchmark = new TcpAsyncPinnedProducer(globalCliArgs, pinnedProducerCliArgs);
-            benchmark.provisionResources();
-            benchmark.run();
+            new TcpAsyncPinnedProducer(globalCliArgs, pinnedProducerCliArgs);
 
             return ExitCode.OK;
         } catch (RuntimeException exception) {

foreign/java/bench/src/main/java/org/apache/iggy/bench/common/enums/ActorKind.java

@@ -0,0 +1,41 @@
+/*
+ * 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.iggy.bench.common.enums;
+
+public enum ActorKind {
+    PRODUCER("producer"),
+    CONSUMER("consumer"),
+    PRODUCING_CONSUMER("producing_consumer");
+
+    private final String value;
+
+    ActorKind(String value) {
+        this.value = value;
+    }
+
+    public String value() {
+        return value;
+    }
+
+    @Override
+    public String toString() {
+        return value;
+    }
+}

foreign/java/bench/src/main/java/org/apache/iggy/bench/common/enums/BenchmarkKind.java

@@ -0,0 +1,46 @@
+/*
+ * 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.iggy.bench.common.enums;
+
+public enum BenchmarkKind {
+    PINNED_PRODUCER("pinned_producer"),
+    PINNED_CONSUMER("pinned_consumer"),
+    PINNED_PRODUCER_AND_CONSUMER("pinned_producer_and_consumer"),
+    BALANCED_PRODUCER("balanced_producer"),
+    BALANCED_CONSUMER_GROUP("balanced_consumer_group"),
+    BALANCED_PRODUCER_AND_CONSUMER_GROUP("balanced_producer_and_consumer_group"),
+    END_TO_END_PRODUCING_CONSUMER("end_to_end_producing_consumer"),
+    END_TO_END_PRODUCING_CONSUMER_GROUP("end_to_end_producing_consumer_group");
+
+    private final String value;
+
+    BenchmarkKind(String value) {
+        this.value = value;
+    }
+
+    public String value() {
+        return value;
+    }
+
+    @Override
+    public String toString() {
+        return value;
+    }
+}