kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cmcc...@apache.org
Subject [kafka] branch trunk updated: MINOR: add useConfiguredPartitioner and skipFlush options for ProduceBench
Date Thu, 04 Jul 2019 00:23:52 GMT
This is an automated email from the ASF dual-hosted git repository.

cmccabe pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 442d362  MINOR: add useConfiguredPartitioner and skipFlush options for ProduceBench
442d362 is described below

commit 442d36241bea6e83c75ed7513e48099b8d73ef39
Author: jolshan <jolshan@confluent.io>
AuthorDate: Wed Jul 3 17:23:37 2019 -0700

    MINOR: add useConfiguredPartitioner and skipFlush options for ProduceBench
    
    Add a "useConfiguredPartitioner" boolean to specify testing with the configured partitioner,
rather than overriding the partitioner in the test.
    
    Add a "skipFlush" boolean to specify skipping the flush operation when producing.  This
is helpful when testing some scenarios where linger.ms is greater than 0.
    
    Reviewers: Colin P. McCabe <cmccabe@apache.org>
---
 .../kafka/trogdor/workload/ProduceBenchSpec.java       | 18 +++++++++++++++++-
 .../kafka/trogdor/workload/ProduceBenchWorker.java     | 16 +++++++++++++---
 .../kafka/trogdor/common/JsonSerializationTest.java    |  2 +-
 3 files changed, 31 insertions(+), 5 deletions(-)

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 34b5393..9f6a907 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
@@ -71,6 +71,8 @@ public class ProduceBenchSpec extends TaskSpec {
     private final Map<String, String> commonClientConf;
     private final TopicsSpec activeTopics;
     private final TopicsSpec inactiveTopics;
+    private final boolean useConfiguredPartitioner;
+    private final boolean skipFlush;
 
     @JsonCreator
     public ProduceBenchSpec(@JsonProperty("startMs") long startMs,
@@ -86,7 +88,9 @@ public class ProduceBenchSpec extends TaskSpec {
                          @JsonProperty("commonClientConf") Map<String, String> commonClientConf,
                          @JsonProperty("adminClientConf") Map<String, String> adminClientConf,
                          @JsonProperty("activeTopics") TopicsSpec activeTopics,
-                         @JsonProperty("inactiveTopics") TopicsSpec inactiveTopics) {
+                         @JsonProperty("inactiveTopics") TopicsSpec inactiveTopics,
+                         @JsonProperty("useConfiguredPartitioner") boolean useConfiguredPartitioner,

+                         @JsonProperty("skipFlush") boolean skipFlush) {
         super(startMs, durationMs);
         this.producerNode = (producerNode == null) ? "" : producerNode;
         this.bootstrapServers = (bootstrapServers == null) ? "" : bootstrapServers;
@@ -104,6 +108,8 @@ public class ProduceBenchSpec extends TaskSpec {
             TopicsSpec.EMPTY : activeTopics.immutableCopy();
         this.inactiveTopics = (inactiveTopics == null) ?
             TopicsSpec.EMPTY : inactiveTopics.immutableCopy();
+        this.useConfiguredPartitioner = useConfiguredPartitioner;
+        this.skipFlush = skipFlush;
     }
 
     @JsonProperty
@@ -166,6 +172,16 @@ public class ProduceBenchSpec extends TaskSpec {
         return inactiveTopics;
     }
 
+    @JsonProperty
+    public boolean useConfiguredPartitioner() {
+        return useConfiguredPartitioner;
+    }
+
+    @JsonProperty
+    public boolean skipFlush() {
+        return skipFlush;
+    }
+
     @Override
     public TaskController newController(String id) {
         return topology -> Collections.singleton(producerNode);
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 84b94d5..9b5159a 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
@@ -214,7 +214,11 @@ public class ProduceBenchWorker implements TaskWorker {
             this.producer = new KafkaProducer<>(props, new ByteArraySerializer(), new
ByteArraySerializer());
             this.keys = new PayloadIterator(spec.keyGenerator());
             this.values = new PayloadIterator(spec.valueGenerator());
-            this.throttle = new SendRecordsThrottle(perPeriod, producer);
+            if (spec.skipFlush()) {
+                this.throttle = new Throttle(perPeriod, THROTTLE_PERIOD_MS);
+            } else {
+                this.throttle = new SendRecordsThrottle(perPeriod, producer);
+            }
         }
 
         @Override
@@ -289,8 +293,14 @@ public class ProduceBenchWorker implements TaskWorker {
                 partitionsIterator = activePartitions.iterator();
 
             TopicPartition partition = partitionsIterator.next();
-            ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(
-                partition.topic(), partition.partition(), keys.next(), values.next());
+            ProducerRecord<byte[], byte[]> record;
+            if (spec.useConfiguredPartitioner()) {
+                record = new ProducerRecord<>(
+                    partition.topic(), keys.next(), values.next());
+            } else {
+                record = new ProducerRecord<>(
+                    partition.topic(), partition.partition(), keys.next(), values.next());
+            }
             sendFuture = producer.send(record,
                 new SendRecordsCallback(this, Time.SYSTEM.milliseconds()));
             throttle.increment();
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 c324ec4..90002ca 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
@@ -55,7 +55,7 @@ public class JsonSerializationTest {
         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, Optional.empty(), null, null, null, null, null));
+            0, 0, null, null, Optional.empty(), null, null, null, null, null, false, false));
         verify(new RoundTripWorkloadSpec(0, 0, null, null, null, null, null, null,
             0, null, null, 0));
         verify(new TopicsSpec());


Mime
View raw message