From f28fed1697a561054a6fcd9d32bf0b9916142717 Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Tue, 6 Nov 2018 13:57:07 +0000 Subject: [PATCH 1/9] KAFKA-7597: Make Trogdor ProduceBenchWorker support transactions It now accepts a new "messagesPerTransaction" field, which, if present, will enable transactional functionality in the bench worker. The producer will open N transactions with X messages each (bounded by the mandatory "maxMessages" field) --- .../trogdor/produce_bench_workload.py | 3 +- .../tests/core/produce_bench_test.py | 25 +++++- .../trogdor/workload/ProduceBenchSpec.java | 41 +++++++++ .../trogdor/workload/ProduceBenchWorker.java | 85 ++++++++++++++----- 4 files changed, 129 insertions(+), 25 deletions(-) diff --git a/tests/kafkatest/services/trogdor/produce_bench_workload.py b/tests/kafkatest/services/trogdor/produce_bench_workload.py index cf6a962b05582..190d6e10fcf14 100644 --- a/tests/kafkatest/services/trogdor/produce_bench_workload.py +++ b/tests/kafkatest/services/trogdor/produce_bench_workload.py @@ -21,12 +21,13 @@ class ProduceBenchWorkloadSpec(TaskSpec): def __init__(self, start_ms, duration_ms, producer_node, bootstrap_servers, target_messages_per_sec, max_messages, producer_conf, admin_client_conf, - common_client_conf, inactive_topics, active_topics): + common_client_conf, inactive_topics, active_topics, messages_per_transaction=0): super(ProduceBenchWorkloadSpec, self).__init__(start_ms, duration_ms) self.message["class"] = "org.apache.kafka.trogdor.workload.ProduceBenchSpec" self.message["producerNode"] = producer_node self.message["bootstrapServers"] = bootstrap_servers self.message["targetMessagesPerSec"] = target_messages_per_sec + self.message["messagesPerTransaction"] = messages_per_transaction self.message["maxMessages"] = max_messages self.message["producerConf"] = producer_conf self.message["adminClientConf"] = admin_client_conf diff --git a/tests/kafkatest/tests/core/produce_bench_test.py b/tests/kafkatest/tests/core/produce_bench_test.py index 125ee941eb5ed..9044cc4389e47 100644 --- a/tests/kafkatest/tests/core/produce_bench_test.py +++ b/tests/kafkatest/tests/core/produce_bench_test.py @@ -31,6 +31,8 @@ def __init__(self, test_context): self.workload_service = ProduceBenchWorkloadService(test_context, self.kafka) self.trogdor = TrogdorService(context=self.test_context, client_services=[self.kafka, self.workload_service]) + self.active_topics = {"produce_bench_topic[0-1]": {"numPartitions": 1, "replicationFactor": 3}} + self.inactive_topics = {"produce_bench_topic[2-9]": {"numPartitions": 1, "replicationFactor": 3}} def setUp(self): self.trogdor.start() @@ -43,8 +45,6 @@ def teardown(self): self.zk.stop() def test_produce_bench(self): - active_topics={"produce_bench_topic[0-1]":{"numPartitions":1, "replicationFactor":3}} - inactive_topics={"produce_bench_topic[2-9]":{"numPartitions":1, "replicationFactor":3}} spec = ProduceBenchWorkloadSpec(0, TaskSpec.MAX_DURATION_MS, self.workload_service.producer_node, self.workload_service.bootstrap_servers, @@ -53,8 +53,25 @@ def test_produce_bench(self): producer_conf={}, admin_client_conf={}, common_client_conf={}, - inactive_topics=inactive_topics, - active_topics=active_topics) + inactive_topics=self.inactive_topics, + active_topics=self.active_topics) + workload1 = self.trogdor.create_task("workload1", spec) + workload1.wait_for_done(timeout_sec=360) + tasks = self.trogdor.tasks() + self.logger.info("TASKS: %s\n" % json.dumps(tasks, sort_keys=True, indent=2)) + + def test_produce_bench_transactions(self): + spec = ProduceBenchWorkloadSpec(0, TaskSpec.MAX_DURATION_MS, + self.workload_service.producer_node, + self.workload_service.bootstrap_servers, + target_messages_per_sec=1000, + max_messages=100000, + producer_conf={}, + admin_client_conf={}, + common_client_conf={}, + inactive_topics=self.inactive_topics, + active_topics=self.active_topics, + messages_per_transaction=10000) # 10 transactions with 10k messages workload1 = self.trogdor.create_task("workload1", spec) workload1.wait_for_done(timeout_sec=360) tasks = self.trogdor.tasks() diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java index 30878bf303f38..fb586253da920 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java @@ -30,12 +30,41 @@ /** * The specification for a benchmark that produces messages to a set of topics. + * + * If the "messagesPerTransaction" field is present, the producer will group and produce messages as separate transactions using the producer transactional API. + * + * An example JSON representation which will result in a producer that creates three topics (foo1, foo2, foo3) + * with three partitions each and produces to them: + * #{@code + * { + * "class": "org.apache.kafka.trogdor.workload.ProduceBenchSpec", + * "durationMs": 10000000, + * "producerNode": "node0", + * "bootstrapServers": "localhost:9092", + * "targetMessagesPerSec": 10, + * "maxMessages": 100, + * "activeTopics": { + * "foo[1-3]": { + * "numPartitions": 3, + * "replicationFactor": 1 + * } + * }, + * "inactiveTopics": { + * "foo[4-5]": { + * "numPartitions": 3, + * "replicationFactor": 1 + * } + * } + * } + * } */ public class ProduceBenchSpec extends TaskSpec { private final String producerNode; private final String bootstrapServers; private final int targetMessagesPerSec; + private final int messagesPerTransaction; private final int maxMessages; + private final boolean useTransactions; private final PayloadGenerator keyGenerator; private final PayloadGenerator valueGenerator; private final Map producerConf; @@ -50,6 +79,7 @@ public ProduceBenchSpec(@JsonProperty("startMs") long startMs, @JsonProperty("producerNode") String producerNode, @JsonProperty("bootstrapServers") String bootstrapServers, @JsonProperty("targetMessagesPerSec") int targetMessagesPerSec, + @JsonProperty("messagesPerTransaction") int messagesPerTransaction, @JsonProperty("maxMessages") int maxMessages, @JsonProperty("keyGenerator") PayloadGenerator keyGenerator, @JsonProperty("valueGenerator") PayloadGenerator valueGenerator, @@ -62,6 +92,8 @@ public ProduceBenchSpec(@JsonProperty("startMs") long startMs, this.producerNode = (producerNode == null) ? "" : producerNode; this.bootstrapServers = (bootstrapServers == null) ? "" : bootstrapServers; this.targetMessagesPerSec = targetMessagesPerSec; + this.useTransactions = messagesPerTransaction != 0; + this.messagesPerTransaction = messagesPerTransaction; this.maxMessages = maxMessages; this.keyGenerator = keyGenerator == null ? new SequentialPayloadGenerator(4, 0) : keyGenerator; @@ -91,6 +123,15 @@ public int targetMessagesPerSec() { return targetMessagesPerSec; } + @JsonProperty + public int messagesPerTransaction() { + return messagesPerTransaction; + } + + public boolean useTransactions() { + return useTransactions; + } + @JsonProperty public int maxMessages() { return maxMessages; diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java index dc749eb65a48f..a8bad513b418f 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java @@ -43,6 +43,7 @@ import java.util.HashSet; import java.util.Iterator; import java.util.Map; +import java.util.UUID; import java.util.Properties; import java.util.concurrent.Callable; import java.util.concurrent.Executors; @@ -50,6 +51,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; public class ProduceBenchWorker implements TaskWorker { private static final Logger log = LoggerFactory.getLogger(ProduceBenchWorker.class); @@ -181,16 +183,23 @@ public class SendRecords implements Callable { private final Throttle throttle; + private Iterator partitionsIterator; + private Future sendFuture; + private AtomicInteger transactionsCommitted; + SendRecords(HashSet activePartitions) { this.activePartitions = activePartitions; + this.partitionsIterator = activePartitions.iterator(); this.histogram = new Histogram(5000); + this.transactionsCommitted = new AtomicInteger(); int perPeriod = WorkerUtils.perSecToPerPeriod(spec.targetMessagesPerSec(), THROTTLE_PERIOD_MS); this.statusUpdaterFuture = executor.scheduleWithFixedDelay( - new StatusUpdater(histogram), 30, 30, TimeUnit.SECONDS); + new StatusUpdater(histogram, transactionsCommitted), 30, 30, TimeUnit.SECONDS); Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, spec.bootstrapServers()); - // add common client configs to producer properties, and then user-specified producer - // configs + if (spec.useTransactions()) + props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction-id-" + UUID.randomUUID()); + // add common client configs to producer properties, and then user-specified producer configs WorkerUtils.addConfigsToProperties(props, spec.commonClientConf(), spec.producerConf()); this.producer = new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer()); this.keys = new PayloadIterator(spec.keyGenerator()); @@ -202,23 +211,26 @@ public class SendRecords implements Callable { public Void call() throws Exception { long startTimeMs = Time.SYSTEM.milliseconds(); try { - Future future = null; try { - Iterator iter = activePartitions.iterator(); - for (int m = 0; m < spec.maxMessages(); m++) { - if (!iter.hasNext()) { - iter = activePartitions.iterator(); + if (spec.useTransactions()) { + producer.initTransactions(); + + for (int m = 0; m < spec.maxMessages() / spec.messagesPerTransaction(); m++) { + sendMessages(spec.messagesPerTransaction()); } - TopicPartition partition = iter.next(); - ProducerRecord record = new ProducerRecord<>( - partition.topic(), partition.partition(), keys.next(), values.next()); - future = producer.send(record, - new SendRecordsCallback(this, Time.SYSTEM.milliseconds())); - throttle.increment(); + int leftOver = spec.maxMessages() % spec.messagesPerTransaction(); + if (leftOver > 0) + sendMessages(leftOver); + } else { + sendMessages(spec.maxMessages()); } + } catch (Exception e) { + if (spec.useTransactions()) + producer.abortTransaction(); + throw e; } finally { - if (future != null) { - future.get(); + if (sendFuture != null) { + sendFuture.get(); } producer.close(); } @@ -226,7 +238,7 @@ public Void call() throws Exception { WorkerUtils.abort(log, "SendRecords", e, doneFuture); } finally { statusUpdaterFuture.cancel(false); - StatusData statusData = new StatusUpdater(histogram).update(); + StatusData statusData = new StatusUpdater(histogram, transactionsCommitted).update(); long curTimeMs = Time.SYSTEM.milliseconds(); log.info("Sent {} total record(s) in {} ms. status: {}", histogram.summarize().numSamples(), curTimeMs - startTimeMs, statusData); @@ -235,6 +247,28 @@ public Void call() throws Exception { return null; } + private void sendMessages(int messageCount) throws InterruptedException { + if (spec.useTransactions()) + producer.beginTransaction(); + + for (int i = 0; i < messageCount; i++) { + if (!partitionsIterator.hasNext()) + partitionsIterator = activePartitions.iterator(); + + TopicPartition partition = partitionsIterator.next(); + ProducerRecord record = new ProducerRecord<>( + partition.topic(), partition.partition(), keys.next(), values.next()); + sendFuture = producer.send(record, + new SendRecordsCallback(this, Time.SYSTEM.milliseconds())); + throttle.increment(); + } + + if (spec.useTransactions()) { + producer.commitTransaction(); + transactionsCommitted.getAndIncrement(); + } + } + void recordDuration(long durationMs) { histogram.add(durationMs); } @@ -242,9 +276,11 @@ void recordDuration(long durationMs) { public class StatusUpdater implements Runnable { private final Histogram histogram; + private final AtomicInteger transactionsCommitted; - StatusUpdater(Histogram histogram) { + StatusUpdater(Histogram histogram, AtomicInteger transactionsCommitted) { this.histogram = histogram; + this.transactionsCommitted = transactionsCommitted; } @Override @@ -261,7 +297,8 @@ StatusData update() { StatusData statusData = new StatusData(summary.numSamples(), summary.average(), summary.percentiles().get(0).value(), summary.percentiles().get(1).value(), - summary.percentiles().get(2).value()); + summary.percentiles().get(2).value(), + transactionsCommitted.get()); status.update(JsonUtil.JSON_SERDE.valueToTree(statusData)); return statusData; } @@ -273,6 +310,7 @@ public static class StatusData { private final int p50LatencyMs; private final int p95LatencyMs; private final int p99LatencyMs; + private final int transactionsCommitted ; /** * The percentiles to use when calculating the histogram data. @@ -285,12 +323,14 @@ public static class StatusData { @JsonProperty("averageLatencyMs") float averageLatencyMs, @JsonProperty("p50LatencyMs") int p50latencyMs, @JsonProperty("p95LatencyMs") int p95latencyMs, - @JsonProperty("p99LatencyMs") int p99latencyMs) { + @JsonProperty("p99LatencyMs") int p99latencyMs, + @JsonProperty("transactionsCommitted") int transactionsCommitted) { this.totalSent = totalSent; this.averageLatencyMs = averageLatencyMs; this.p50LatencyMs = p50latencyMs; this.p95LatencyMs = p95latencyMs; this.p99LatencyMs = p99latencyMs; + this.transactionsCommitted = transactionsCommitted; } @JsonProperty @@ -298,6 +338,11 @@ public long totalSent() { return totalSent; } + @JsonProperty + public long transactionsCommitted() { + return transactionsCommitted; + } + @JsonProperty public float averageLatencyMs() { return averageLatencyMs; From 6013d79f9b38ddf1f1ab9a24b7a3d3d4861c36fb Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Tue, 20 Nov 2018 11:43:33 +0000 Subject: [PATCH 2/9] Drive produceBench transactions through a new TransactionGenerator field --- ...trogdor-run-transactional-produce-bench.sh | 51 +++++++++++ .../trogdor/produce_bench_workload.py | 7 +- .../tests/core/produce_bench_test.py | 6 +- .../trogdor/workload/ProduceBenchSpec.java | 26 +++--- .../trogdor/workload/ProduceBenchWorker.java | 84 ++++++++++++------- .../workload/TransactionActionGenerator.java | 47 +++++++++++ .../UniformTransactionsGenerator.java | 62 ++++++++++++++ .../workload/ZeroTransactionsGenerator.java | 29 +++++++ 8 files changed, 266 insertions(+), 46 deletions(-) create mode 100755 tests/bin/trogdor-run-transactional-produce-bench.sh create mode 100644 tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java create mode 100644 tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java create mode 100644 tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java diff --git a/tests/bin/trogdor-run-transactional-produce-bench.sh b/tests/bin/trogdor-run-transactional-produce-bench.sh new file mode 100755 index 0000000000000..fd5ff0a01f215 --- /dev/null +++ b/tests/bin/trogdor-run-transactional-produce-bench.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash +# 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. + +COORDINATOR_ENDPOINT="localhost:8889" +TASK_ID="produce_bench_$RANDOM" +TASK_SPEC=$( +cat < producerConf; private final Map adminClientConf; private final Map commonClientConf; @@ -79,10 +79,10 @@ public ProduceBenchSpec(@JsonProperty("startMs") long startMs, @JsonProperty("producerNode") String producerNode, @JsonProperty("bootstrapServers") String bootstrapServers, @JsonProperty("targetMessagesPerSec") int targetMessagesPerSec, - @JsonProperty("messagesPerTransaction") int messagesPerTransaction, @JsonProperty("maxMessages") int maxMessages, @JsonProperty("keyGenerator") PayloadGenerator keyGenerator, @JsonProperty("valueGenerator") PayloadGenerator valueGenerator, + @JsonProperty("transactionGenerator") TransactionActionGenerator txGenerator, @JsonProperty("producerConf") Map producerConf, @JsonProperty("commonClientConf") Map commonClientConf, @JsonProperty("adminClientConf") Map adminClientConf, @@ -92,13 +92,13 @@ public ProduceBenchSpec(@JsonProperty("startMs") long startMs, this.producerNode = (producerNode == null) ? "" : producerNode; this.bootstrapServers = (bootstrapServers == null) ? "" : bootstrapServers; this.targetMessagesPerSec = targetMessagesPerSec; - this.useTransactions = messagesPerTransaction != 0; - this.messagesPerTransaction = messagesPerTransaction; this.maxMessages = maxMessages; this.keyGenerator = keyGenerator == null ? new SequentialPayloadGenerator(4, 0) : keyGenerator; this.valueGenerator = valueGenerator == null ? new ConstantPayloadGenerator(512, new byte[0]) : valueGenerator; + this.transactionGenerator = txGenerator == null ? + new ZeroTransactionsGenerator() : txGenerator; this.producerConf = configOrEmptyMap(producerConf); this.commonClientConf = configOrEmptyMap(commonClientConf); this.adminClientConf = configOrEmptyMap(adminClientConf); @@ -123,15 +123,6 @@ public int targetMessagesPerSec() { return targetMessagesPerSec; } - @JsonProperty - public int messagesPerTransaction() { - return messagesPerTransaction; - } - - public boolean useTransactions() { - return useTransactions; - } - @JsonProperty public int maxMessages() { return maxMessages; @@ -147,6 +138,11 @@ public PayloadGenerator valueGenerator() { return valueGenerator; } + @JsonProperty + public TransactionActionGenerator transactionGenerator() { + return transactionGenerator; + } + @JsonProperty public Map producerConf() { return producerConf; diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java index a8bad513b418f..9061fc4ba5dfc 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java @@ -36,6 +36,7 @@ import org.apache.kafka.trogdor.common.WorkerUtils; import org.apache.kafka.trogdor.task.TaskWorker; import org.apache.kafka.trogdor.task.WorkerStatusTracker; +import org.apache.kafka.trogdor.workload.TransactionActionGenerator.TransactionActions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -181,23 +182,32 @@ public class SendRecords implements Callable { private final PayloadIterator values; + private final TransactionActionGenerator txActionGenerator; + private final Throttle throttle; private Iterator partitionsIterator; private Future sendFuture; private AtomicInteger transactionsCommitted; + private boolean toUseTransactions = false; SendRecords(HashSet activePartitions) { this.activePartitions = activePartitions; this.partitionsIterator = activePartitions.iterator(); this.histogram = new Histogram(5000); + + this.txActionGenerator = spec.transactionGenerator(); this.transactionsCommitted = new AtomicInteger(); + if (txActionGenerator.nextAction() == TransactionActions.INIT_TRANSACTIONS) + toUseTransactions = true; + int perPeriod = WorkerUtils.perSecToPerPeriod(spec.targetMessagesPerSec(), THROTTLE_PERIOD_MS); this.statusUpdaterFuture = executor.scheduleWithFixedDelay( new StatusUpdater(histogram, transactionsCommitted), 30, 30, TimeUnit.SECONDS); + Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, spec.bootstrapServers()); - if (spec.useTransactions()) + if (toUseTransactions) props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction-id-" + UUID.randomUUID()); // add common client configs to producer properties, and then user-specified producer configs WorkerUtils.addConfigsToProperties(props, spec.commonClientConf(), spec.producerConf()); @@ -212,20 +222,22 @@ public Void call() throws Exception { long startTimeMs = Time.SYSTEM.milliseconds(); try { try { - if (spec.useTransactions()) { + if (toUseTransactions) producer.initTransactions(); - for (int m = 0; m < spec.maxMessages() / spec.messagesPerTransaction(); m++) { - sendMessages(spec.messagesPerTransaction()); + int sentMessages = 0; + while (sentMessages < spec.maxMessages()) { + if (toUseTransactions) { + boolean tookAction = takeTransactionAction(); + if (tookAction) + continue; } - int leftOver = spec.maxMessages() % spec.messagesPerTransaction(); - if (leftOver > 0) - sendMessages(leftOver); - } else { - sendMessages(spec.maxMessages()); + sendMessage(); + sentMessages++; } + takeTransactionAction(); // give the transactionGenerator a chance to commit if configured evenly } catch (Exception e) { - if (spec.useTransactions()) + if (toUseTransactions) producer.abortTransaction(); throw e; } finally { @@ -247,26 +259,42 @@ public Void call() throws Exception { return null; } - private void sendMessages(int messageCount) throws InterruptedException { - if (spec.useTransactions()) - producer.beginTransaction(); - - for (int i = 0; i < messageCount; i++) { - if (!partitionsIterator.hasNext()) - partitionsIterator = activePartitions.iterator(); - - TopicPartition partition = partitionsIterator.next(); - ProducerRecord record = new ProducerRecord<>( - partition.topic(), partition.partition(), keys.next(), values.next()); - sendFuture = producer.send(record, - new SendRecordsCallback(this, Time.SYSTEM.milliseconds())); - throttle.increment(); + private boolean takeTransactionAction() { + boolean tookAction = true; + TransactionActions nextAction = txActionGenerator.nextAction(); + switch (nextAction) { + case INIT_TRANSACTIONS: + throw new IllegalStateException("Cannot initiate transactions twice"); + case BEGIN_TRANSACTION: + log.debug("Beginning transaction."); + producer.beginTransaction(); + break; + case COMMIT_TRANSACTION: + log.debug("Committing transaction."); + producer.commitTransaction(); + transactionsCommitted.getAndIncrement(); + break; + case ABORT_TRANSACTION: + log.debug("Aborting transaction."); + producer.abortTransaction(); + break; + case NO_OP: + tookAction = false; + break; } + return tookAction; + } - if (spec.useTransactions()) { - producer.commitTransaction(); - transactionsCommitted.getAndIncrement(); - } + private void sendMessage() throws InterruptedException { + if (!partitionsIterator.hasNext()) + partitionsIterator = activePartitions.iterator(); + + TopicPartition partition = partitionsIterator.next(); + ProducerRecord record = new ProducerRecord<>( + partition.topic(), partition.partition(), keys.next(), values.next()); + sendFuture = producer.send(record, + new SendRecordsCallback(this, Time.SYSTEM.milliseconds())); + throttle.increment(); } void recordDuration(long durationMs) { diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java new file mode 100644 index 0000000000000..8590b8e5bb65a --- /dev/null +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java @@ -0,0 +1,47 @@ +/* + * 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.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; + +/** + * Generates actions that should be taken by a producer that uses transactions. + */ +@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, + include = JsonTypeInfo.As.PROPERTY, + property = "type") +@JsonSubTypes(value = { + @JsonSubTypes.Type(value = ZeroTransactionsGenerator.class, name = "zeroTransactions"), + @JsonSubTypes.Type(value = UniformTransactionsGenerator.class, name = "uniform"), +}) +public interface TransactionActionGenerator { + enum TransactionActions { + INIT_TRANSACTIONS, BEGIN_TRANSACTION, COMMIT_TRANSACTION, ABORT_TRANSACTION, NO_OP + } + + /** + * Returns the next action that the producer should take in regards to transactions. + * This method should be called every time before a producer sends a message. + * This means that most of the time it should return #{@link TransactionActions#NO_OP} + * to signal the producer that its next step should be to send a message. + * + * The first returned action of every generator should be #{@link TransactionActions#INIT_TRANSACTIONS}, + * in order to signal to the producer that it should be configured for using transactions + */ + TransactionActions nextAction(); +} diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java new file mode 100644 index 0000000000000..fe7bf9e8443c7 --- /dev/null +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java @@ -0,0 +1,62 @@ +/* + * 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; + +/** + * A uniform transactions generator where every N records are grouped in a separate transaction + */ +public class UniformTransactionsGenerator implements TransactionActionGenerator { + + private final int messagesPerTransaction; + private boolean initialized = false; + private int messagesInTransaction = -1; + + @JsonCreator + public UniformTransactionsGenerator(@JsonProperty("messagesPerTransaction") int messagesPerTransaction) { + if (messagesPerTransaction < 1) + throw new IllegalArgumentException("Cannot have less than one message per transaction."); + + this.messagesPerTransaction = messagesPerTransaction; + } + + @JsonProperty + public int messagesPerTransaction() { + return messagesPerTransaction; + } + + @Override + public TransactionActions nextAction() { + if (!initialized) { + initialized = true; + return TransactionActions.INIT_TRANSACTIONS; + } + if (messagesInTransaction == -1) { + messagesInTransaction = 0; + return TransactionActions.BEGIN_TRANSACTION; + } + if (messagesInTransaction == messagesPerTransaction) { + messagesInTransaction = -1; + return TransactionActions.COMMIT_TRANSACTION; + } + + messagesInTransaction += 1; + return TransactionActions.NO_OP; + } +} diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java new file mode 100644 index 0000000000000..36fe232bc101e --- /dev/null +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java @@ -0,0 +1,29 @@ +/* + * 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; + +public class ZeroTransactionsGenerator implements TransactionActionGenerator { + @JsonCreator + public ZeroTransactionsGenerator() {} + + @Override + public TransactionActions nextAction() { + return TransactionActions.NO_OP; + } +} From df9cede8197773cf20b20f70821193e69d3c8b25 Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Wed, 21 Nov 2018 07:47:13 +0000 Subject: [PATCH 3/9] Address PR comments --- .../kafka/trogdor/workload/ProduceBenchWorker.java | 6 +++--- .../trogdor/workload/TransactionActionGenerator.java | 10 +++++----- .../trogdor/workload/UniformTransactionsGenerator.java | 10 +++++----- .../trogdor/workload/ZeroTransactionsGenerator.java | 4 ++-- 4 files changed, 15 insertions(+), 15 deletions(-) diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java index 9061fc4ba5dfc..a6f909eedeb68 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java @@ -36,7 +36,7 @@ import org.apache.kafka.trogdor.common.WorkerUtils; import org.apache.kafka.trogdor.task.TaskWorker; import org.apache.kafka.trogdor.task.WorkerStatusTracker; -import org.apache.kafka.trogdor.workload.TransactionActionGenerator.TransactionActions; +import org.apache.kafka.trogdor.workload.TransactionActionGenerator.TransactionAction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -198,7 +198,7 @@ public class SendRecords implements Callable { this.txActionGenerator = spec.transactionGenerator(); this.transactionsCommitted = new AtomicInteger(); - if (txActionGenerator.nextAction() == TransactionActions.INIT_TRANSACTIONS) + if (txActionGenerator.nextAction() == TransactionAction.INIT_TRANSACTIONS) toUseTransactions = true; int perPeriod = WorkerUtils.perSecToPerPeriod(spec.targetMessagesPerSec(), THROTTLE_PERIOD_MS); @@ -261,7 +261,7 @@ public Void call() throws Exception { private boolean takeTransactionAction() { boolean tookAction = true; - TransactionActions nextAction = txActionGenerator.nextAction(); + TransactionAction nextAction = txActionGenerator.nextAction(); switch (nextAction) { case INIT_TRANSACTIONS: throw new IllegalStateException("Cannot initiate transactions twice"); diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java index 8590b8e5bb65a..93cf52a3f284b 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java @@ -26,22 +26,22 @@ include = JsonTypeInfo.As.PROPERTY, property = "type") @JsonSubTypes(value = { - @JsonSubTypes.Type(value = ZeroTransactionsGenerator.class, name = "zeroTransactions"), + @JsonSubTypes.Type(value = ZeroTransactionsGenerator.class, name = "zero"), @JsonSubTypes.Type(value = UniformTransactionsGenerator.class, name = "uniform"), }) public interface TransactionActionGenerator { - enum TransactionActions { + enum TransactionAction { INIT_TRANSACTIONS, BEGIN_TRANSACTION, COMMIT_TRANSACTION, ABORT_TRANSACTION, NO_OP } /** * Returns the next action that the producer should take in regards to transactions. * This method should be called every time before a producer sends a message. - * This means that most of the time it should return #{@link TransactionActions#NO_OP} + * This means that most of the time it should return #{@link TransactionAction#NO_OP} * to signal the producer that its next step should be to send a message. * - * The first returned action of every generator should be #{@link TransactionActions#INIT_TRANSACTIONS}, + * The first returned action of every generator should be #{@link TransactionAction#INIT_TRANSACTIONS}, * in order to signal to the producer that it should be configured for using transactions */ - TransactionActions nextAction(); + TransactionAction nextAction(); } diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java index fe7bf9e8443c7..804e757ababc2 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java @@ -42,21 +42,21 @@ public int messagesPerTransaction() { } @Override - public TransactionActions nextAction() { + public synchronized TransactionAction nextAction() { if (!initialized) { initialized = true; - return TransactionActions.INIT_TRANSACTIONS; + return TransactionAction.INIT_TRANSACTIONS; } if (messagesInTransaction == -1) { messagesInTransaction = 0; - return TransactionActions.BEGIN_TRANSACTION; + return TransactionAction.BEGIN_TRANSACTION; } if (messagesInTransaction == messagesPerTransaction) { messagesInTransaction = -1; - return TransactionActions.COMMIT_TRANSACTION; + return TransactionAction.COMMIT_TRANSACTION; } messagesInTransaction += 1; - return TransactionActions.NO_OP; + return TransactionAction.NO_OP; } } diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java index 36fe232bc101e..6b2414172c195 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java @@ -23,7 +23,7 @@ public class ZeroTransactionsGenerator implements TransactionActionGenerator { public ZeroTransactionsGenerator() {} @Override - public TransactionActions nextAction() { - return TransactionActions.NO_OP; + public TransactionAction nextAction() { + return TransactionAction.NO_OP; } } From e96469c59dcd0e5f5a760a40ece6181c8b2b84cf Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Wed, 21 Nov 2018 20:15:38 +0000 Subject: [PATCH 4/9] Renames --- .../apache/kafka/trogdor/workload/ProduceBenchSpec.java | 8 ++++---- .../apache/kafka/trogdor/workload/ProduceBenchWorker.java | 8 ++++---- ...tionActionGenerator.java => TransactionGenerator.java} | 4 ++-- .../trogdor/workload/UniformTransactionsGenerator.java | 4 ++-- .../kafka/trogdor/workload/ZeroTransactionsGenerator.java | 4 ++-- 5 files changed, 14 insertions(+), 14 deletions(-) rename tools/src/main/java/org/apache/kafka/trogdor/workload/{TransactionActionGenerator.java => TransactionGenerator.java} (96%) diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java index 26bd3aae29efd..97227defc713a 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java @@ -31,7 +31,7 @@ /** * The specification for a benchmark that produces messages to a set of topics. * - * To configure a transactional producer, a #{@link TransactionActionGenerator} must be passed in. + * To configure a transactional producer, a #{@link TransactionGenerator} must be passed in. * Said generator works in lockstep with the producer by instructing it what action to take next in regards to a transaction. * * An example JSON representation which will result in a producer that creates three topics (foo1, foo2, foo3) @@ -66,7 +66,7 @@ public class ProduceBenchSpec extends TaskSpec { private final int maxMessages; private final PayloadGenerator keyGenerator; private final PayloadGenerator valueGenerator; - private final TransactionActionGenerator transactionGenerator; + private final TransactionGenerator transactionGenerator; private final Map producerConf; private final Map adminClientConf; private final Map commonClientConf; @@ -82,7 +82,7 @@ public ProduceBenchSpec(@JsonProperty("startMs") long startMs, @JsonProperty("maxMessages") int maxMessages, @JsonProperty("keyGenerator") PayloadGenerator keyGenerator, @JsonProperty("valueGenerator") PayloadGenerator valueGenerator, - @JsonProperty("transactionGenerator") TransactionActionGenerator txGenerator, + @JsonProperty("transactionGenerator") TransactionGenerator txGenerator, @JsonProperty("producerConf") Map producerConf, @JsonProperty("commonClientConf") Map commonClientConf, @JsonProperty("adminClientConf") Map adminClientConf, @@ -139,7 +139,7 @@ public PayloadGenerator valueGenerator() { } @JsonProperty - public TransactionActionGenerator transactionGenerator() { + public TransactionGenerator transactionGenerator() { return transactionGenerator; } diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java index a6f909eedeb68..4ee333ebf12bb 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java @@ -36,7 +36,7 @@ import org.apache.kafka.trogdor.common.WorkerUtils; import org.apache.kafka.trogdor.task.TaskWorker; import org.apache.kafka.trogdor.task.WorkerStatusTracker; -import org.apache.kafka.trogdor.workload.TransactionActionGenerator.TransactionAction; +import org.apache.kafka.trogdor.workload.TransactionGenerator.TransactionAction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -182,7 +182,7 @@ public class SendRecords implements Callable { private final PayloadIterator values; - private final TransactionActionGenerator txActionGenerator; + private final TransactionGenerator txActionGenerator; private final Throttle throttle; @@ -198,7 +198,7 @@ public class SendRecords implements Callable { this.txActionGenerator = spec.transactionGenerator(); this.transactionsCommitted = new AtomicInteger(); - if (txActionGenerator.nextAction() == TransactionAction.INIT_TRANSACTIONS) + if (txActionGenerator.action() == TransactionAction.INIT_TRANSACTIONS) toUseTransactions = true; int perPeriod = WorkerUtils.perSecToPerPeriod(spec.targetMessagesPerSec(), THROTTLE_PERIOD_MS); @@ -261,7 +261,7 @@ public Void call() throws Exception { private boolean takeTransactionAction() { boolean tookAction = true; - TransactionAction nextAction = txActionGenerator.nextAction(); + TransactionAction nextAction = txActionGenerator.action(); switch (nextAction) { case INIT_TRANSACTIONS: throw new IllegalStateException("Cannot initiate transactions twice"); diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java similarity index 96% rename from tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java rename to tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java index 93cf52a3f284b..99a9759a18823 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionActionGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java @@ -29,7 +29,7 @@ @JsonSubTypes.Type(value = ZeroTransactionsGenerator.class, name = "zero"), @JsonSubTypes.Type(value = UniformTransactionsGenerator.class, name = "uniform"), }) -public interface TransactionActionGenerator { +public interface TransactionGenerator { enum TransactionAction { INIT_TRANSACTIONS, BEGIN_TRANSACTION, COMMIT_TRANSACTION, ABORT_TRANSACTION, NO_OP } @@ -43,5 +43,5 @@ enum TransactionAction { * The first returned action of every generator should be #{@link TransactionAction#INIT_TRANSACTIONS}, * in order to signal to the producer that it should be configured for using transactions */ - TransactionAction nextAction(); + TransactionAction action(); } diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java index 804e757ababc2..8f79074d4366c 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java @@ -22,7 +22,7 @@ /** * A uniform transactions generator where every N records are grouped in a separate transaction */ -public class UniformTransactionsGenerator implements TransactionActionGenerator { +public class UniformTransactionsGenerator implements TransactionGenerator { private final int messagesPerTransaction; private boolean initialized = false; @@ -42,7 +42,7 @@ public int messagesPerTransaction() { } @Override - public synchronized TransactionAction nextAction() { + public synchronized TransactionAction action() { if (!initialized) { initialized = true; return TransactionAction.INIT_TRANSACTIONS; diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java index 6b2414172c195..1306d3247a8f6 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java @@ -18,12 +18,12 @@ import com.fasterxml.jackson.annotation.JsonCreator; -public class ZeroTransactionsGenerator implements TransactionActionGenerator { +public class ZeroTransactionsGenerator implements TransactionGenerator { @JsonCreator public ZeroTransactionsGenerator() {} @Override - public TransactionAction nextAction() { + public TransactionAction action() { return TransactionAction.NO_OP; } } From 64574ac6df5a695f15ee3e3a2a731eb7012bdec7 Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Mon, 26 Nov 2018 21:12:08 +0000 Subject: [PATCH 5/9] Rename variable and checkstyle fix --- .../kafka/trogdor/workload/ProduceBenchWorker.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java index 4ee333ebf12bb..7b834fcb0d1f8 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java @@ -182,7 +182,7 @@ public class SendRecords implements Callable { private final PayloadIterator values; - private final TransactionGenerator txActionGenerator; + private final TransactionGenerator transactionGenerator; private final Throttle throttle; @@ -196,9 +196,9 @@ public class SendRecords implements Callable { this.partitionsIterator = activePartitions.iterator(); this.histogram = new Histogram(5000); - this.txActionGenerator = spec.transactionGenerator(); + this.transactionGenerator = spec.transactionGenerator(); this.transactionsCommitted = new AtomicInteger(); - if (txActionGenerator.action() == TransactionAction.INIT_TRANSACTIONS) + if (transactionGenerator.action() == TransactionAction.INIT_TRANSACTIONS) toUseTransactions = true; int perPeriod = WorkerUtils.perSecToPerPeriod(spec.targetMessagesPerSec(), THROTTLE_PERIOD_MS); @@ -261,7 +261,7 @@ public Void call() throws Exception { private boolean takeTransactionAction() { boolean tookAction = true; - TransactionAction nextAction = txActionGenerator.action(); + TransactionAction nextAction = transactionGenerator.action(); switch (nextAction) { case INIT_TRANSACTIONS: throw new IllegalStateException("Cannot initiate transactions twice"); @@ -338,7 +338,7 @@ public static class StatusData { private final int p50LatencyMs; private final int p95LatencyMs; private final int p99LatencyMs; - private final int transactionsCommitted ; + private final int transactionsCommitted; /** * The percentiles to use when calculating the histogram data. From 37ec9e038f00626c3a71142c0c690ce02f9a125d Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Mon, 26 Nov 2018 22:29:24 +0000 Subject: [PATCH 6/9] Remove zeroTransactionGenerator class and make use of Optionals jackson vars Imports new library - jackson-datatype-jdk8. This should be used until jackson 3 comes out, at which point we can switch to it completely Removed INIT_TRANSACTIONS state from the `TransactionAction` enum, this eases TransactionGenerator implementations Removed ZeroTransactionsGenerator class - it is much more intuitive to provide null or not define the transactionsGenerator class --- build.gradle | 1 + gradle/dependencies.gradle | 1 + .../trogdor/produce_bench_workload.py | 2 -- .../apache/kafka/trogdor/common/JsonUtil.java | 2 ++ .../trogdor/workload/ProduceBenchSpec.java | 10 +++---- .../trogdor/workload/ProduceBenchWorker.java | 15 +++++----- .../workload/TransactionGenerator.java | 6 +--- .../UniformTransactionsGenerator.java | 5 ---- .../workload/ZeroTransactionsGenerator.java | 29 ------------------- .../trogdor/common/JsonSerializationTest.java | 3 +- 10 files changed, 19 insertions(+), 55 deletions(-) delete mode 100644 tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java diff --git a/build.gradle b/build.gradle index 4d514dfcf97f0..8f3ec6bde545c 100644 --- a/build.gradle +++ b/build.gradle @@ -918,6 +918,7 @@ project(':tools') { compile project(':log4j-appender') compile libs.argparse4j compile libs.jacksonDatabind + compile libs.jacksonDatatype compile libs.slf4jApi compile libs.jacksonJaxrsJsonProvider diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle index 7dd3604db089c..6f1395436f129 100644 --- a/gradle/dependencies.gradle +++ b/gradle/dependencies.gradle @@ -103,6 +103,7 @@ libs += [ bcpkix: "org.bouncycastle:bcpkix-jdk15on:$versions.bcpkix", easymock: "org.easymock:easymock:$versions.easymock", jacksonDatabind: "com.fasterxml.jackson.core:jackson-databind:$versions.jackson", + jacksonDatatype: "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:$versions.jackson", jacksonJaxrsJsonProvider: "com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:$versions.jackson", jaxbApi: "javax.xml.bind:jaxb-api:$versions.jaxb", jaxrsApi: "javax.ws.rs:javax.ws.rs-api:$versions.jaxrs", diff --git a/tests/kafkatest/services/trogdor/produce_bench_workload.py b/tests/kafkatest/services/trogdor/produce_bench_workload.py index ade2d91055d9a..9afc81462aedf 100644 --- a/tests/kafkatest/services/trogdor/produce_bench_workload.py +++ b/tests/kafkatest/services/trogdor/produce_bench_workload.py @@ -24,8 +24,6 @@ def __init__(self, start_ms, duration_ms, producer_node, bootstrap_servers, common_client_conf, inactive_topics, active_topics, transaction_generator=None): super(ProduceBenchWorkloadSpec, self).__init__(start_ms, duration_ms) - if transaction_generator is None: - transaction_generator = {"type": "zeroTransactions"} self.message["class"] = "org.apache.kafka.trogdor.workload.ProduceBenchSpec" self.message["producerNode"] = producer_node self.message["bootstrapServers"] = bootstrap_servers diff --git a/tools/src/main/java/org/apache/kafka/trogdor/common/JsonUtil.java b/tools/src/main/java/org/apache/kafka/trogdor/common/JsonUtil.java index 70193c3beefb1..ad90ffc6e8422 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/common/JsonUtil.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/common/JsonUtil.java @@ -22,6 +22,7 @@ import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; /** * Utilities for working with JSON. @@ -33,6 +34,7 @@ public class JsonUtil { JSON_SERDE = new ObjectMapper(); JSON_SERDE.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false); JSON_SERDE.configure(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY, true); + JSON_SERDE.registerModule(new Jdk8Module()); JSON_SERDE.setSerializationInclusion(JsonInclude.Include.NON_EMPTY); } diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java index 97227defc713a..30b7db5e25051 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java @@ -26,6 +26,7 @@ import java.util.Collections; import java.util.Map; +import java.util.Optional; import java.util.Set; /** @@ -66,7 +67,7 @@ public class ProduceBenchSpec extends TaskSpec { private final int maxMessages; private final PayloadGenerator keyGenerator; private final PayloadGenerator valueGenerator; - private final TransactionGenerator transactionGenerator; + private final Optional transactionGenerator; private final Map producerConf; private final Map adminClientConf; private final Map commonClientConf; @@ -82,7 +83,7 @@ public ProduceBenchSpec(@JsonProperty("startMs") long startMs, @JsonProperty("maxMessages") int maxMessages, @JsonProperty("keyGenerator") PayloadGenerator keyGenerator, @JsonProperty("valueGenerator") PayloadGenerator valueGenerator, - @JsonProperty("transactionGenerator") TransactionGenerator txGenerator, + @JsonProperty("transactionGenerator") Optional txGenerator, @JsonProperty("producerConf") Map producerConf, @JsonProperty("commonClientConf") Map commonClientConf, @JsonProperty("adminClientConf") Map adminClientConf, @@ -97,8 +98,7 @@ public ProduceBenchSpec(@JsonProperty("startMs") long startMs, new SequentialPayloadGenerator(4, 0) : keyGenerator; this.valueGenerator = valueGenerator == null ? new ConstantPayloadGenerator(512, new byte[0]) : valueGenerator; - this.transactionGenerator = txGenerator == null ? - new ZeroTransactionsGenerator() : txGenerator; + this.transactionGenerator = txGenerator; this.producerConf = configOrEmptyMap(producerConf); this.commonClientConf = configOrEmptyMap(commonClientConf); this.adminClientConf = configOrEmptyMap(adminClientConf); @@ -139,7 +139,7 @@ public PayloadGenerator valueGenerator() { } @JsonProperty - public TransactionGenerator transactionGenerator() { + public Optional transactionGenerator() { return transactionGenerator; } diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java index 7b834fcb0d1f8..4b65fd3b0cee7 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java @@ -44,8 +44,9 @@ import java.util.HashSet; import java.util.Iterator; import java.util.Map; -import java.util.UUID; +import java.util.Optional; import java.util.Properties; +import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -182,7 +183,7 @@ public class SendRecords implements Callable { private final PayloadIterator values; - private final TransactionGenerator transactionGenerator; + private final Optional transactionGenerator; private final Throttle throttle; @@ -197,9 +198,8 @@ public class SendRecords implements Callable { this.histogram = new Histogram(5000); this.transactionGenerator = spec.transactionGenerator(); + this.toUseTransactions = this.transactionGenerator.isPresent(); this.transactionsCommitted = new AtomicInteger(); - if (transactionGenerator.action() == TransactionAction.INIT_TRANSACTIONS) - toUseTransactions = true; int perPeriod = WorkerUtils.perSecToPerPeriod(spec.targetMessagesPerSec(), THROTTLE_PERIOD_MS); this.statusUpdaterFuture = executor.scheduleWithFixedDelay( @@ -235,7 +235,8 @@ public Void call() throws Exception { sendMessage(); sentMessages++; } - takeTransactionAction(); // give the transactionGenerator a chance to commit if configured evenly + if (toUseTransactions) + takeTransactionAction(); // give the transactionGenerator a chance to commit if configured evenly } catch (Exception e) { if (toUseTransactions) producer.abortTransaction(); @@ -261,10 +262,8 @@ public Void call() throws Exception { private boolean takeTransactionAction() { boolean tookAction = true; - TransactionAction nextAction = transactionGenerator.action(); + TransactionAction nextAction = transactionGenerator.get().action(); switch (nextAction) { - case INIT_TRANSACTIONS: - throw new IllegalStateException("Cannot initiate transactions twice"); case BEGIN_TRANSACTION: log.debug("Beginning transaction."); producer.beginTransaction(); diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java index 99a9759a18823..bd98444919d75 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java @@ -26,12 +26,11 @@ include = JsonTypeInfo.As.PROPERTY, property = "type") @JsonSubTypes(value = { - @JsonSubTypes.Type(value = ZeroTransactionsGenerator.class, name = "zero"), @JsonSubTypes.Type(value = UniformTransactionsGenerator.class, name = "uniform"), }) public interface TransactionGenerator { enum TransactionAction { - INIT_TRANSACTIONS, BEGIN_TRANSACTION, COMMIT_TRANSACTION, ABORT_TRANSACTION, NO_OP + BEGIN_TRANSACTION, COMMIT_TRANSACTION, ABORT_TRANSACTION, NO_OP } /** @@ -39,9 +38,6 @@ enum TransactionAction { * This method should be called every time before a producer sends a message. * This means that most of the time it should return #{@link TransactionAction#NO_OP} * to signal the producer that its next step should be to send a message. - * - * The first returned action of every generator should be #{@link TransactionAction#INIT_TRANSACTIONS}, - * in order to signal to the producer that it should be configured for using transactions */ TransactionAction action(); } diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java index 8f79074d4366c..f1d99f039b166 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java @@ -25,7 +25,6 @@ public class UniformTransactionsGenerator implements TransactionGenerator { private final int messagesPerTransaction; - private boolean initialized = false; private int messagesInTransaction = -1; @JsonCreator @@ -43,10 +42,6 @@ public int messagesPerTransaction() { @Override public synchronized TransactionAction action() { - if (!initialized) { - initialized = true; - return TransactionAction.INIT_TRANSACTIONS; - } if (messagesInTransaction == -1) { messagesInTransaction = 0; return TransactionAction.BEGIN_TRANSACTION; diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java deleted file mode 100644 index 1306d3247a8f6..0000000000000 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ZeroTransactionsGenerator.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * 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; - -public class ZeroTransactionsGenerator implements TransactionGenerator { - @JsonCreator - public ZeroTransactionsGenerator() {} - - @Override - public TransactionAction action() { - return TransactionAction.NO_OP; - } -} diff --git a/tools/src/test/java/org/apache/kafka/trogdor/common/JsonSerializationTest.java b/tools/src/test/java/org/apache/kafka/trogdor/common/JsonSerializationTest.java index 5e6ff81e28b4d..c324ec4c70872 100644 --- a/tools/src/test/java/org/apache/kafka/trogdor/common/JsonSerializationTest.java +++ b/tools/src/test/java/org/apache/kafka/trogdor/common/JsonSerializationTest.java @@ -37,6 +37,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import static org.junit.Assert.assertNotNull; @@ -54,7 +55,7 @@ public void testDeserializationDoesNotProduceNulls() throws Exception { verify(new WorkerRunning(null, null, 0, null)); verify(new WorkerStopping(null, null, 0, null)); verify(new ProduceBenchSpec(0, 0, null, null, - 0, 0, null, null, null, null, null, null, null)); + 0, 0, null, null, Optional.empty(), null, null, null, null, null)); verify(new RoundTripWorkloadSpec(0, 0, null, null, null, null, null, null, 0, null, null, 0)); verify(new TopicsSpec()); From f17afb9952b3bf37554790bd54ab559edd53dffc Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Tue, 27 Nov 2018 17:02:41 +0000 Subject: [PATCH 7/9] Rename jackson jdk8 lib name and add to all projects that use jackson --- build.gradle | 6 +++++- gradle/dependencies.gradle | 2 +- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/build.gradle b/build.gradle index 8f3ec6bde545c..5ce648adf776f 100644 --- a/build.gradle +++ b/build.gradle @@ -565,6 +565,7 @@ project(':core') { dependencies { compile project(':clients') compile libs.jacksonDatabind + compile libs.jacksonJDK8Datatypes compile libs.joptSimple compile libs.metrics compile libs.scalaLibrary @@ -830,6 +831,7 @@ project(':clients') { compile libs.snappy compile libs.slf4jApi compileOnly libs.jacksonDatabind // for SASL/OAUTHBEARER bearer token parsing + compileOnly libs.jacksonJDK8Datatypes jacksonDatabindConfig libs.jacksonDatabind // to publish as provided scope dependency. @@ -839,6 +841,7 @@ project(':clients') { testRuntime libs.slf4jlog4j testRuntime libs.jacksonDatabind + testRuntime libs.jacksonJDK8Datatypes } task determineCommitId { @@ -918,7 +921,7 @@ project(':tools') { compile project(':log4j-appender') compile libs.argparse4j compile libs.jacksonDatabind - compile libs.jacksonDatatype + compile libs.jacksonJDK8Datatypes compile libs.slf4jApi compile libs.jacksonJaxrsJsonProvider @@ -1348,6 +1351,7 @@ project(':connect:json') { dependencies { compile project(':connect:api') compile libs.jacksonDatabind + compile libs.jacksonJDK8Datatypes compile libs.slf4jApi testCompile libs.easymock diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle index 6f1395436f129..59f56fcd4aba4 100644 --- a/gradle/dependencies.gradle +++ b/gradle/dependencies.gradle @@ -103,7 +103,7 @@ libs += [ bcpkix: "org.bouncycastle:bcpkix-jdk15on:$versions.bcpkix", easymock: "org.easymock:easymock:$versions.easymock", jacksonDatabind: "com.fasterxml.jackson.core:jackson-databind:$versions.jackson", - jacksonDatatype: "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:$versions.jackson", + jacksonJDK8Datatypes: "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:$versions.jackson", jacksonJaxrsJsonProvider: "com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:$versions.jackson", jaxbApi: "javax.xml.bind:jaxb-api:$versions.jaxb", jaxrsApi: "javax.ws.rs:javax.ws.rs-api:$versions.jaxrs", From 9591d55153b36519861db857226684181303736c Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Tue, 27 Nov 2018 18:23:59 +0000 Subject: [PATCH 8/9] Rename txGenerator#action() method, use long for transactionsCommitted var, rename transactional id --- .../trogdor/workload/ProduceBenchWorker.java | 32 +++++++++---------- .../workload/TransactionGenerator.java | 2 +- .../UniformTransactionsGenerator.java | 2 +- 3 files changed, 18 insertions(+), 18 deletions(-) diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java index 4b65fd3b0cee7..abf5976892120 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchWorker.java @@ -53,7 +53,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; public class ProduceBenchWorker implements TaskWorker { private static final Logger log = LoggerFactory.getLogger(ProduceBenchWorker.class); @@ -189,8 +189,8 @@ public class SendRecords implements Callable { private Iterator partitionsIterator; private Future sendFuture; - private AtomicInteger transactionsCommitted; - private boolean toUseTransactions = false; + private AtomicLong transactionsCommitted; + private boolean enableTransactions; SendRecords(HashSet activePartitions) { this.activePartitions = activePartitions; @@ -198,8 +198,8 @@ public class SendRecords implements Callable { this.histogram = new Histogram(5000); this.transactionGenerator = spec.transactionGenerator(); - this.toUseTransactions = this.transactionGenerator.isPresent(); - this.transactionsCommitted = new AtomicInteger(); + this.enableTransactions = this.transactionGenerator.isPresent(); + this.transactionsCommitted = new AtomicLong(); int perPeriod = WorkerUtils.perSecToPerPeriod(spec.targetMessagesPerSec(), THROTTLE_PERIOD_MS); this.statusUpdaterFuture = executor.scheduleWithFixedDelay( @@ -207,8 +207,8 @@ public class SendRecords implements Callable { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, spec.bootstrapServers()); - if (toUseTransactions) - props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction-id-" + UUID.randomUUID()); + if (enableTransactions) + props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "produce-bench-transaction-id-" + UUID.randomUUID()); // add common client configs to producer properties, and then user-specified producer configs WorkerUtils.addConfigsToProperties(props, spec.commonClientConf(), spec.producerConf()); this.producer = new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer()); @@ -222,12 +222,12 @@ public Void call() throws Exception { long startTimeMs = Time.SYSTEM.milliseconds(); try { try { - if (toUseTransactions) + if (enableTransactions) producer.initTransactions(); int sentMessages = 0; while (sentMessages < spec.maxMessages()) { - if (toUseTransactions) { + if (enableTransactions) { boolean tookAction = takeTransactionAction(); if (tookAction) continue; @@ -235,10 +235,10 @@ public Void call() throws Exception { sendMessage(); sentMessages++; } - if (toUseTransactions) + if (enableTransactions) takeTransactionAction(); // give the transactionGenerator a chance to commit if configured evenly } catch (Exception e) { - if (toUseTransactions) + if (enableTransactions) producer.abortTransaction(); throw e; } finally { @@ -262,7 +262,7 @@ public Void call() throws Exception { private boolean takeTransactionAction() { boolean tookAction = true; - TransactionAction nextAction = transactionGenerator.get().action(); + TransactionAction nextAction = transactionGenerator.get().nextAction(); switch (nextAction) { case BEGIN_TRANSACTION: log.debug("Beginning transaction."); @@ -303,9 +303,9 @@ void recordDuration(long durationMs) { public class StatusUpdater implements Runnable { private final Histogram histogram; - private final AtomicInteger transactionsCommitted; + private final AtomicLong transactionsCommitted; - StatusUpdater(Histogram histogram, AtomicInteger transactionsCommitted) { + StatusUpdater(Histogram histogram, AtomicLong transactionsCommitted) { this.histogram = histogram; this.transactionsCommitted = transactionsCommitted; } @@ -337,7 +337,7 @@ public static class StatusData { private final int p50LatencyMs; private final int p95LatencyMs; private final int p99LatencyMs; - private final int transactionsCommitted; + private final long transactionsCommitted; /** * The percentiles to use when calculating the histogram data. @@ -351,7 +351,7 @@ public static class StatusData { @JsonProperty("p50LatencyMs") int p50latencyMs, @JsonProperty("p95LatencyMs") int p95latencyMs, @JsonProperty("p99LatencyMs") int p99latencyMs, - @JsonProperty("transactionsCommitted") int transactionsCommitted) { + @JsonProperty("transactionsCommitted") long transactionsCommitted) { this.totalSent = totalSent; this.averageLatencyMs = averageLatencyMs; this.p50LatencyMs = p50latencyMs; diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java index bd98444919d75..5ec47ec91c167 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/TransactionGenerator.java @@ -39,5 +39,5 @@ enum TransactionAction { * This means that most of the time it should return #{@link TransactionAction#NO_OP} * to signal the producer that its next step should be to send a message. */ - TransactionAction action(); + TransactionAction nextAction(); } diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java index f1d99f039b166..1fbfbc233818f 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/UniformTransactionsGenerator.java @@ -41,7 +41,7 @@ public int messagesPerTransaction() { } @Override - public synchronized TransactionAction action() { + public synchronized TransactionAction nextAction() { if (messagesInTransaction == -1) { messagesInTransaction = 0; return TransactionAction.BEGIN_TRANSACTION; From 7fa5340f39aaed9662a5a94e2ddff0e377d0b358 Mon Sep 17 00:00:00 2001 From: Stanislav Kozlovski Date: Tue, 27 Nov 2018 18:29:10 +0000 Subject: [PATCH 9/9] Check for null when given transactionGenerator in ProduceBenchSpec --- .../org/apache/kafka/trogdor/workload/ProduceBenchSpec.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java index 30b7db5e25051..c0bbd7eb0120b 100644 --- a/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java +++ b/tools/src/main/java/org/apache/kafka/trogdor/workload/ProduceBenchSpec.java @@ -98,7 +98,7 @@ public ProduceBenchSpec(@JsonProperty("startMs") long startMs, new SequentialPayloadGenerator(4, 0) : keyGenerator; this.valueGenerator = valueGenerator == null ? new ConstantPayloadGenerator(512, new byte[0]) : valueGenerator; - this.transactionGenerator = txGenerator; + this.transactionGenerator = txGenerator == null ? Optional.empty() : txGenerator; this.producerConf = configOrEmptyMap(producerConf); this.commonClientConf = configOrEmptyMap(commonClientConf); this.adminClientConf = configOrEmptyMap(adminClientConf);