KAFKA-8102: Add an interval-based Trogdor transaction generator (#6444)

This patch adds a TimeIntervalTransactionsGenerator class which enables the Trogdor ProduceBench worker to commit transactions based on a configurable millisecond time interval.

Also, we now handle 409 create task responses in the coordinator command-line client by printing a more informative message

Reviewers: Colin P. McCabe <[email protected]>

Author: stanislavkozlovski

TROGDOR.md

@@ -105,6 +105,7 @@ Trogdor can run several workloads.  Workloads perform operations on the cluster
 
 ### ProduceBench
 ProduceBench starts a Kafka producer on a single agent node, producing to several partitions.  The workload measures the average produce latency, as well as the median, 95th percentile, and 99th percentile latency.
+It can be configured to use a transactional producer which can commit transactions based on a set time interval or number of messages.
 
 ### RoundTripWorkload
 RoundTripWorkload tests both production and consumption.  The workload starts a Kafka producer and consumer on a single node.  The consumer will read back the messages that were produced by the producer.

tests/spec/transactional-produce-bench.json

@@ -15,7 +15,7 @@
 
 //
 // An example task specification for running a transactional producer benchmark
-in Trogdor.  See TROGDOR.md for details.
+// in Trogdor.  See TROGDOR.md for details.
 //
 
 {

tools/src/main/java/org/apache/kafka/trogdor/coordinator/CoordinatorClient.java

@@ -44,6 +44,7 @@
 import org.apache.kafka.trogdor.rest.TaskState;
 import org.apache.kafka.trogdor.rest.TasksResponse;
 import org.apache.kafka.trogdor.task.TaskSpec;
+import org.apache.kafka.trogdor.rest.RequestConflictException;
 import org.apache.kafka.trogdor.rest.UptimeResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -426,8 +427,15 @@ public static void main(String[] args) throws Exception {
                 TaskSpec taskSpec = JsonUtil.
                     objectFromCommandLineArgument(res.getString("taskSpec"), TaskSpec.class);
                 CreateTaskRequest req = new CreateTaskRequest(taskId, taskSpec);
-                client.createTask(req);
-                System.out.printf("Sent CreateTaskRequest for task %s.%n", req.id());
+                try {
+                    client.createTask(req);
+                    System.out.printf("Sent CreateTaskRequest for task %s.%n", req.id());
+                } catch (RequestConflictException rce) {
+                    System.out.printf("CreateTaskRequest for task %s got a 409 status code - " +
+                        "a task with the same ID but a different specification already exists.%nException: %s%n",
+                        req.id(), rce.getMessage());
+                    Exit.exit(1);
+                }
                 break;
             }
             case "stopTask": {

tools/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java

@@ -302,6 +302,7 @@ TreeMap<String, Long> activeWorkerIds() {
      *
      * @param id                    The ID of the task to create.
      * @param spec                  The specification of the task to create.
+     * @throws RequestConflictException - if a task with the same ID but different spec exists
      */
     public void createTask(final String id, TaskSpec spec)
             throws Throwable {

tools/src/main/java/org/apache/kafka/trogdor/workload/TimeIntervalTransactionsGenerator.java

@@ -0,0 +1,67 @@
+/*
+ * 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.kafka.trogdor.workload;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.kafka.common.utils.Time;
+
+/**
+ * A transactions generator where we commit a transaction every N milliseconds
+ */
+public class TimeIntervalTransactionsGenerator implements TransactionGenerator {
+
+    private static final long NULL_START_MS = -1;
+
+    private final Time time;
+    private final int intervalMs;
+
+    private long lastTransactionStartMs = NULL_START_MS;
+
+    @JsonCreator
+    public TimeIntervalTransactionsGenerator(@JsonProperty("transactionIntervalMs") int intervalMs) {
+        this(intervalMs, Time.SYSTEM);
+    }
+
+    TimeIntervalTransactionsGenerator(@JsonProperty("transactionIntervalMs") int intervalMs,
+                                      Time time) {
+        if (intervalMs < 1) {
+            throw new IllegalArgumentException("Cannot have a negative interval");
+        }
+        this.time = time;
+        this.intervalMs = intervalMs;
+    }
+
+    @JsonProperty
+    public int transactionIntervalMs() {
+        return intervalMs;
+    }
+
+    @Override
+    public synchronized TransactionAction nextAction() {
+        if (lastTransactionStartMs == NULL_START_MS) {
+            lastTransactionStartMs = time.milliseconds();
+            return TransactionAction.BEGIN_TRANSACTION;
+        }
+        if (time.milliseconds() - lastTransactionStartMs >= intervalMs) {
+            lastTransactionStartMs = NULL_START_MS;
+            return TransactionAction.COMMIT_TRANSACTION;
+        }
+
+        return TransactionAction.NO_OP;
+    }
+}

tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java

@@ -27,6 +27,7 @@
     property = "type")
 @JsonSubTypes(value = {
     @JsonSubTypes.Type(value = UniformTransactionsGenerator.class, name = "uniform"),
+    @JsonSubTypes.Type(value = TimeIntervalTransactionsGenerator.class, name = "interval"),
 })
 public interface TransactionGenerator {
     enum TransactionAction {

tools/src/test/java/org/apache/kafka/trogdor/workload/TimeIntervalTransactionsGeneratorTest.java

@@ -0,0 +1,42 @@
+/*
+ * 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.kafka.trogdor.workload;
+
+import org.apache.kafka.common.utils.MockTime;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class TimeIntervalTransactionsGeneratorTest {
+    @Test
+    public void testCommitsTransactionAfterIntervalPasses() {
+        MockTime time = new MockTime();
+        TimeIntervalTransactionsGenerator generator = new TimeIntervalTransactionsGenerator(100, time);
+
+        assertEquals(100, generator.transactionIntervalMs());
+        assertEquals(TransactionGenerator.TransactionAction.BEGIN_TRANSACTION, generator.nextAction());
+        assertEquals(TransactionGenerator.TransactionAction.NO_OP, generator.nextAction());
+        time.sleep(50);
+        assertEquals(TransactionGenerator.TransactionAction.NO_OP, generator.nextAction());
+        time.sleep(49);
+        assertEquals(TransactionGenerator.TransactionAction.NO_OP, generator.nextAction());
+        time.sleep(1);
+        assertEquals(TransactionGenerator.TransactionAction.COMMIT_TRANSACTION, generator.nextAction());
+        assertEquals(TransactionGenerator.TransactionAction.BEGIN_TRANSACTION, generator.nextAction());
+    }
+}