You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "brandboat (via GitHub)" <gi...@apache.org> on 2024/07/21 16:30:45 UTC
[PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
brandboat opened a new pull request, #16646:
URL: https://github.com/apache/kafka/pull/16646
related to https://issues.apache.org/jira/browse/KAFKA-16900,
Currently, users need to set --transaction-duration-ms to enable transactions in kafka-producer-perf-test, which is not straightforward. A better approach is to enable transactions when a transaction ID is provided.
This PR allows enabling transaction in kafka-producer-perf-test by either
- set `transaction.id=<id>` via `--producer-props` or
- set `transaction.id=<id>` in config file via `--producer.config` or
- set `--transaction-id <id>` or
- set `--transaction-duration-ms=<ms>`
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16646:
URL: https://github.com/apache/kafka/pull/16646#issuecomment-2287092644
> The sad thing is that I still can't make all system tests that related to ProducerPerformance passed.
Could you share the details to me ? I run the e2e on trunk, all pass
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16646:
URL: https://github.com/apache/kafka/pull/16646#discussion_r1695591590
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -494,4 +473,60 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
}
}
+ static final class ConfigPostProcessor {
+ final String topicName;
+ final long numRecords;
+ final Integer recordSize;
+ final double throughput;
+ final boolean payloadMonotonic;
+ final Properties producerProps;
+ final boolean shouldPrintMetrics;
+ final Long transactionDurationMs;
+ final boolean transactionsEnabled;
+ final List<byte[]> payloadByteList;
+
+ public ConfigPostProcessor(ArgumentParser parser, String[] args) throws IOException, ArgumentParserException {
+ Namespace namespace = parser.parseArgs(args);
+ this.topicName = namespace.getString("topic");
+ this.numRecords = namespace.getLong("numRecords");
+ this.recordSize = namespace.getInt("recordSize");
+ this.throughput = namespace.getDouble("throughput");
+ this.payloadMonotonic = namespace.getBoolean("payloadMonotonic");
+ this.shouldPrintMetrics = namespace.getBoolean("printMetrics");
+
+ List<String> producerConfigs = namespace.getList("producerConfig");
+ String producerConfigFile = namespace.getString("producerConfigFile");
+ String payloadFilePath = namespace.getString("payloadFile");
+ Long transactionDurationMsArg = namespace.getLong("transactionDurationMs");
+ String transactionIdArg = namespace.getString("transactionalId");
+ if (producerConfigs == null && producerConfigFile == null) {
+ throw new ArgumentParserException("Either --producer-props or --producer.config must be specified.", parser);
+ }
+ if (transactionDurationMsArg != null && transactionDurationMsArg <= 0) {
+ throw new ArgumentParserException("--transaction-duration-ms should > 0", parser);
+ }
+
+ // since default value gets printed with the help text, we are escaping \n there and replacing it with correct value here.
+ String payloadDelimiter = namespace.getString("payloadDelimiter").equals("\\n")
+ ? "\n" : namespace.getString("payloadDelimiter");
+ this.payloadByteList = readPayloadFile(payloadFilePath, payloadDelimiter);
+ this.producerProps = readProps(producerConfigs, producerConfigFile);
+ // setup transaction related configs
+ this.transactionsEnabled = transactionDurationMsArg != null
+ || transactionIdArg != null
+ || producerProps.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
+ if (transactionsEnabled) {
+ Optional<String> txIdInProps =
+ Optional.ofNullable(producerProps.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG))
+ .map(Object::toString);
+ String transactionId = Optional.ofNullable(transactionIdArg).orElse(txIdInProps.orElse(DEFAULT_TRANSACTION_ID));
Review Comment:
Should we use UUID instead in order to allow users run this perf tool concurrently? Otherwise, using the same default transaction ID will be a issue.
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -47,79 +48,56 @@
public class ProducerPerformance {
+ public static final String DEFAULT_TRANSACTION_ID = "performance-producer-default-transactional-id";
+ public static final Long DEFAULT_TRANSACTION_DURATION_MS = 3000L;
Review Comment:
`long`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16646:
URL: https://github.com/apache/kafka/pull/16646#issuecomment-2282912727
@brandboat any updates?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #16646:
URL: https://github.com/apache/kafka/pull/16646#discussion_r1697116056
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -494,4 +473,60 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
}
}
+ static final class ConfigPostProcessor {
+ final String topicName;
+ final long numRecords;
+ final Integer recordSize;
+ final double throughput;
+ final boolean payloadMonotonic;
+ final Properties producerProps;
+ final boolean shouldPrintMetrics;
+ final Long transactionDurationMs;
+ final boolean transactionsEnabled;
+ final List<byte[]> payloadByteList;
+
+ public ConfigPostProcessor(ArgumentParser parser, String[] args) throws IOException, ArgumentParserException {
+ Namespace namespace = parser.parseArgs(args);
+ this.topicName = namespace.getString("topic");
+ this.numRecords = namespace.getLong("numRecords");
+ this.recordSize = namespace.getInt("recordSize");
+ this.throughput = namespace.getDouble("throughput");
+ this.payloadMonotonic = namespace.getBoolean("payloadMonotonic");
+ this.shouldPrintMetrics = namespace.getBoolean("printMetrics");
+
+ List<String> producerConfigs = namespace.getList("producerConfig");
+ String producerConfigFile = namespace.getString("producerConfigFile");
+ String payloadFilePath = namespace.getString("payloadFile");
+ Long transactionDurationMsArg = namespace.getLong("transactionDurationMs");
+ String transactionIdArg = namespace.getString("transactionalId");
+ if (producerConfigs == null && producerConfigFile == null) {
+ throw new ArgumentParserException("Either --producer-props or --producer.config must be specified.", parser);
+ }
+ if (transactionDurationMsArg != null && transactionDurationMsArg <= 0) {
+ throw new ArgumentParserException("--transaction-duration-ms should > 0", parser);
+ }
+
+ // since default value gets printed with the help text, we are escaping \n there and replacing it with correct value here.
+ String payloadDelimiter = namespace.getString("payloadDelimiter").equals("\\n")
+ ? "\n" : namespace.getString("payloadDelimiter");
+ this.payloadByteList = readPayloadFile(payloadFilePath, payloadDelimiter);
+ this.producerProps = readProps(producerConfigs, producerConfigFile);
+ // setup transaction related configs
+ this.transactionsEnabled = transactionDurationMsArg != null
+ || transactionIdArg != null
+ || producerProps.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
+ if (transactionsEnabled) {
+ Optional<String> txIdInProps =
+ Optional.ofNullable(producerProps.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG))
+ .map(Object::toString);
+ String transactionId = Optional.ofNullable(transactionIdArg).orElse(txIdInProps.orElse(DEFAULT_TRANSACTION_ID));
Review Comment:
Use unique transaction id by default make sense to me, I didn't change this since this is the original implementation see https://issues.apache.org/jira/browse/KAFKA-5491. And I didn't find the reason we use same transaction id in the beginning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16646:
URL: https://github.com/apache/kafka/pull/16646#issuecomment-2289206498
> I got bunch of errors as follows
oh, I run into the same error now. I file https://issues.apache.org/jira/browse/KAFKA-17343 to trace it. Please feel free to take over it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #16646:
URL: https://github.com/apache/kafka/pull/16646#discussion_r1688212531
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -330,18 +346,15 @@ static ArgumentParser argParser() {
.type(String.class)
.metavar("TRANSACTIONAL-ID")
.dest("transactionalId")
- .setDefault("performance-producer-default-transactional-id")
- .help("The transactionalId to use if transaction-duration-ms is > 0. Useful when testing the performance of concurrent transactions.");
+ .help("The transactional id to use. This config take precedence over transactional.id via --producer.config or --producer-props");
Review Comment:
Addressed in the latest commit, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #16646:
URL: https://github.com/apache/kafka/pull/16646#discussion_r1688212021
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -81,7 +84,22 @@ void start(String[] args) throws IOException {
List<byte[]> payloadByteList = readPayloadFile(payloadFilePath, payloadDelimiter);
- Properties props = readProps(producerProps, producerConfig, transactionalId, transactionsEnabled);
+ Properties props = readProps(producerProps, producerConfig);
Review Comment:
I refactored all arguments in ProducerPerformance into a new class to facilitate writing more unit tests and improving test coverage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on PR #16646:
URL: https://github.com/apache/kafka/pull/16646#issuecomment-2284361793
> @brandboat any updates?
The sad thing is that I still can't make all system tests that related to `ProducerPerformance` passed.
```
TC_PATHS="tests/kafkatest/benchmarks/core/benchmark_test.py tests/kafkatest/sanity_checks/test_performance_services.py tests/kafkatest/tests/core/throttling_test.py tests/kafkatest/tests/client/quota_test.py tests/kafkatest/services/performance/producer_performance.py" bash tests/docker/run_tests.sh
```
Most of the failed tests are in benchmark_test.py. Even after checking out the trunk branch, the issue persists.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16646:
URL: https://github.com/apache/kafka/pull/16646#discussion_r1685930254
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -330,18 +346,15 @@ static ArgumentParser argParser() {
.type(String.class)
.metavar("TRANSACTIONAL-ID")
.dest("transactionalId")
- .setDefault("performance-producer-default-transactional-id")
- .help("The transactionalId to use if transaction-duration-ms is > 0. Useful when testing the performance of concurrent transactions.");
+ .help("The transactional id to use. This config take precedence over transactional.id via --producer.config or --producer-props");
Review Comment:
Could you please remind users that `--transaction-duration-ms` will have default value if transaction id is defined.
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -81,7 +84,22 @@ void start(String[] args) throws IOException {
List<byte[]> payloadByteList = readPayloadFile(payloadFilePath, payloadDelimiter);
- Properties props = readProps(producerProps, producerConfig, transactionalId, transactionsEnabled);
+ Properties props = readProps(producerProps, producerConfig);
Review Comment:
Maybe we can enhance `readProps` to handle those changes? With that we can write UT for `readProps` easily.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16646:
URL: https://github.com/apache/kafka/pull/16646#issuecomment-2260242037
@brandboat Could you please run the related py tests on you local?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #16646:
URL: https://github.com/apache/kafka/pull/16646
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on PR #16646:
URL: https://github.com/apache/kafka/pull/16646#issuecomment-2289176368
> Could you share the details to me ? I run the e2e on trunk, all pass
Interesting... After ran `TC_PATHS="tests/kafkatest/benchmarks/core/benchmark_test.py" bash tests/docker/run_tests.sh` in trunk branch, I got bunch of errors as follows
```
[INFO:2024-08-14 08:52:07,138]: Triggering test 2 of 62...
[INFO:2024-08-14 08:52:07,148]: RunnerClient: Loading test {'directory': '/opt/kafka-dev/tests/kafkatest/benchmarks/core', 'file_name': 'benchmark_test.py', 'cls_name': 'Benchmark', 'method_name': 'test_consumer_throughput', 'injected_args': {'security_protocol': 'PLAINTEXT', 'compression_type': 'none'}}
[INFO:2024-08-14 08:52:07,154]: RunnerClient: kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=none: on run 1/1
[INFO:2024-08-14 08:52:07,156]: RunnerClient: kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=none: Setting up...
[INFO:2024-08-14 08:52:09,266]: RunnerClient: kafkatest.benchmarks.core.benchmark_test.Benchmark.test_consumer_throughput.security_protocol=PLAINTEXT.compression_type=none: Running...
Exception in thread ProducerPerformanceService-0-127377594939280-worker-1:
Traceback (most recent call last):
File "/opt/kafka-dev/tests/kafkatest/services/performance/producer_performance.py", line 157, in _worker
self.results[idx-1] = self.parse_stats(last)
File "/opt/kafka-dev/tests/kafkatest/services/performance/producer_performance.py", line 170, in parse_stats
'latency_50th_ms': float(parts[4].split()[0]),
IndexError: list index out of range
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/lib/python3.9/threading.py", line 954, in _bootstrap_inner
self.run()
File "/usr/lib/python3.9/threading.py", line 892, in run
self._target(*self._args, **self._kwargs)
File "/usr/local/lib/python3.9/dist-packages/ducktape/services/background_thread.py", line 38, in _protected_worker
self._worker(idx, node)
File "/opt/kafka-dev/tests/kafkatest/services/performance/producer_performance.py", line 159, in _worker
raise Exception("Unable to parse aggregate performance statistics on node %d: %s" % (idx, last))
Exception: Unable to parse aggregate performance statistics on node 1: 2545045 records sent, 505470.7 records/sec (48.21 MB/sec), 788.0 ms avg latency, 1736.0 ms max latency.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16646:
URL: https://github.com/apache/kafka/pull/16646#discussion_r1689647830
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -494,4 +483,100 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
}
}
+ static final class ConfigPostProcessor {
+ private final String topicName;
+ private final long numRecords;
+ private final Integer recordSize;
+ private final double throughput;
+ private final boolean payloadMonotonic;
+ private final Properties producerProps;
+ private final boolean shouldPrintMetrics;
+ private final Long transactionDurationMs;
+ private final boolean transactionsEnabled;
+ private final List<byte[]> payloadByteList;
+
+ public ConfigPostProcessor(ArgumentParser parser, String[] args) throws IOException, ArgumentParserException {
+ Namespace namespace = parser.parseArgs(args);
+ this.topicName = namespace.getString("topic");
+ this.numRecords = namespace.getLong("numRecords");
+ this.recordSize = namespace.getInt("recordSize");
+ this.throughput = namespace.getDouble("throughput");
+ this.payloadMonotonic = namespace.getBoolean("payloadMonotonic");
+ this.shouldPrintMetrics = namespace.getBoolean("printMetrics");
+
+ List<String> producerConfigs = namespace.getList("producerConfig");
+ String producerConfigFile = namespace.getString("producerConfigFile");
+ String payloadFilePath = namespace.getString("payloadFile");
+ Long transactionDurationMsArg = namespace.getLong("transactionDurationMs");
+ String transactionIdArg = namespace.getString("transactionalId");
+ if (producerConfigs == null && producerConfigFile == null) {
+ throw new ArgumentParserException("Either --producer-props or --producer.config must be specified.", parser);
+ }
+ if (transactionDurationMsArg != null && transactionDurationMsArg <= 0) {
+ throw new ArgumentParserException("--transaction-duration-ms should > 0", parser);
+ }
+
+ // since default value gets printed with the help text, we are escaping \n there and replacing it with correct value here.
+ String payloadDelimiter = namespace.getString("payloadDelimiter").equals("\\n")
+ ? "\n" : namespace.getString("payloadDelimiter");
+ this.payloadByteList = readPayloadFile(payloadFilePath, payloadDelimiter);
+ this.producerProps = readProps(producerConfigs, producerConfigFile);
+ // setup transaction related configs
+ this.transactionsEnabled = transactionDurationMsArg != null
+ || transactionIdArg != null
+ || producerProps.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
+ if (transactionsEnabled) {
+ Optional<String> txIdInProps =
+ Optional.ofNullable(producerProps.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG))
+ .map(Object::toString);
+ String transactionId = Optional.ofNullable(transactionIdArg).orElse(txIdInProps.orElse(DEFAULT_TRANSACTION_ID));
+ producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);
+
+ if (transactionDurationMsArg == null) {
+ transactionDurationMsArg = DEFAULT_TRANSACTION_DURATION_MS;
+ }
+ }
+ this.transactionDurationMs = transactionDurationMsArg;
+ }
+
+ public String topicName() {
Review Comment:
This is temporary class, so we remove those setters and then change all inner variable to be "package-private"
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -47,43 +48,30 @@
public class ProducerPerformance {
+ public static final String DEFAULT_TRANSACTION_ID = "performance-producer-default-transactional-id";
+ public static final Long DEFAULT_TRANSACTION_DURATION_MS = 3000L;
+
public static void main(String[] args) throws Exception {
ProducerPerformance perf = new ProducerPerformance();
perf.start(args);
}
-
+
void start(String[] args) throws IOException {
ArgumentParser parser = argParser();
try {
- Namespace res = parser.parseArgs(args);
-
- /* parse args */
- String topicName = res.getString("topic");
- long numRecords = res.getLong("numRecords");
- Integer recordSize = res.getInt("recordSize");
- double throughput = res.getDouble("throughput");
- boolean payloadMonotonic = res.getBoolean("payloadMonotonic");
- List<String> producerProps = res.getList("producerConfig");
- String producerConfig = res.getString("producerConfigFile");
- String payloadFilePath = res.getString("payloadFile");
- String transactionalId = res.getString("transactionalId");
- boolean shouldPrintMetrics = res.getBoolean("printMetrics");
- long transactionDurationMs = res.getLong("transactionDurationMs");
- boolean transactionsEnabled = 0 < transactionDurationMs;
-
- // since default value gets printed with the help text, we are escaping \n there and replacing it with correct value here.
- String payloadDelimiter = res.getString("payloadDelimiter").equals("\\n") ? "\n" : res.getString("payloadDelimiter");
-
- if (producerProps == null && producerConfig == null) {
- throw new ArgumentParserException("Either --producer-props or --producer.config must be specified.", parser);
- }
-
- List<byte[]> payloadByteList = readPayloadFile(payloadFilePath, payloadDelimiter);
-
- Properties props = readProps(producerProps, producerConfig, transactionalId, transactionsEnabled);
-
- KafkaProducer<byte[], byte[]> producer = createKafkaProducer(props);
+ ConfigPostProcessor config = new ConfigPostProcessor(parser, args);
Review Comment:
Maybe we can remove those local variables? `config.xxx` is good to me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16900 kafka-producer-perf-test reports error when using transaction. [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #16646:
URL: https://github.com/apache/kafka/pull/16646#discussion_r1689883367
##########
tools/src/main/java/org/apache/kafka/tools/ProducerPerformance.java:
##########
@@ -47,43 +48,30 @@
public class ProducerPerformance {
+ public static final String DEFAULT_TRANSACTION_ID = "performance-producer-default-transactional-id";
+ public static final Long DEFAULT_TRANSACTION_DURATION_MS = 3000L;
+
public static void main(String[] args) throws Exception {
ProducerPerformance perf = new ProducerPerformance();
perf.start(args);
}
-
+
void start(String[] args) throws IOException {
ArgumentParser parser = argParser();
try {
- Namespace res = parser.parseArgs(args);
-
- /* parse args */
- String topicName = res.getString("topic");
- long numRecords = res.getLong("numRecords");
- Integer recordSize = res.getInt("recordSize");
- double throughput = res.getDouble("throughput");
- boolean payloadMonotonic = res.getBoolean("payloadMonotonic");
- List<String> producerProps = res.getList("producerConfig");
- String producerConfig = res.getString("producerConfigFile");
- String payloadFilePath = res.getString("payloadFile");
- String transactionalId = res.getString("transactionalId");
- boolean shouldPrintMetrics = res.getBoolean("printMetrics");
- long transactionDurationMs = res.getLong("transactionDurationMs");
- boolean transactionsEnabled = 0 < transactionDurationMs;
-
- // since default value gets printed with the help text, we are escaping \n there and replacing it with correct value here.
- String payloadDelimiter = res.getString("payloadDelimiter").equals("\\n") ? "\n" : res.getString("payloadDelimiter");
-
- if (producerProps == null && producerConfig == null) {
- throw new ArgumentParserException("Either --producer-props or --producer.config must be specified.", parser);
- }
-
- List<byte[]> payloadByteList = readPayloadFile(payloadFilePath, payloadDelimiter);
-
- Properties props = readProps(producerProps, producerConfig, transactionalId, transactionsEnabled);
-
- KafkaProducer<byte[], byte[]> producer = createKafkaProducer(props);
+ ConfigPostProcessor config = new ConfigPostProcessor(parser, args);
Review Comment:
Done, thanks for reviewing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org