You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/03/07 06:08:06 UTC
[GitHub] [kafka] peterwanner opened a new pull request #11855: MINOR: Clarify producer batch.size in upgrade docs
peterwanner opened a new pull request #11855:
URL: https://github.com/apache/kafka/pull/11855
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### 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
[GitHub] [kafka] peterwanner commented on pull request #11855: MINOR: fix default producer batch.size config
Posted by GitBox <gi...@apache.org>.
peterwanner commented on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1065987918
@showuon ok, maybe i mis-understanding what you say in the first comment. Best Thanks for your reminder.
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11855: MINOR: fix default producer batch.size config
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#discussion_r825377329
##########
File path: core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala
##########
@@ -123,4 +163,37 @@ class ConsoleProducerTest {
assertEquals("console-producer",
producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG))
}
+
+ @Test
+ def testBatchSizeOverride(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeOverride)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals("456",
+ producerConfig.getString(ProducerConfig.BATCH_SIZE_CONFIG))
+ }
+
+ @Test
+ def testBatchSizeSet(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeSet)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals("123",
+ producerConfig.getString(ProducerConfig.BATCH_SIZE_CONFIG))
+ }
+
+ @Test
+ def testDefaultBatchSize(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeDefault)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals(16*1024,
+ producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG))
+ }
+
+ @Test
+ def testBatchSizeNotSet(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeNotSet)
Review comment:
I think this test can be renamed to `testBatchSizeNotSetAndMaxPartitionMemoryBytesSet` to be clear.
And also the config variable name can be `batchSizeNotSetAndMaxPartitionMemoryBytesSet`
##########
File path: core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala
##########
@@ -123,4 +163,37 @@ class ConsoleProducerTest {
assertEquals("console-producer",
producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG))
}
+
+ @Test
+ def testBatchSizeOverride(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeOverride)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals("456",
+ producerConfig.getString(ProducerConfig.BATCH_SIZE_CONFIG))
+ }
+
+ @Test
+ def testBatchSizeSet(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeSet)
Review comment:
renamed to `testBatchSizeSetAndMaxPartitionMemoryBytesNotSet`
And also the config variable name `batchSizeSet` -> `batchSizeSetAndMaxPartitionMemoryBytesNotSet`
--
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
[GitHub] [kafka] peterwanner edited a comment on pull request #11855: MINOR: bring producer batch.size back
Posted by GitBox <gi...@apache.org>.
peterwanner edited a comment on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1065832199
@showuon could you look at this pr? or as what you say, it's reaching the end of the release
--
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
[GitHub] [kafka] showuon commented on pull request #11855: MINOR: set `batch-size` option value into batch.size config in consoleProducer
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1066051900
Failed tests are unrelated:
```
Build / JDK 8 and Scala 2.12 / kafka.api.ConsumerBounceTest.testClose()
Build / JDK 17 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testPathToJsonFile, Security=PLAINTEXT
Build / JDK 17 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testPreferredReplicaElection, Security=PLAINTEXT
Build / JDK 17 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testTopicPartition, Security=PLAINTEXT
Build / JDK 17 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testPathToJsonFile, Security=PLAINTEXT
Build / JDK 11 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()
Build / JDK 11 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()
Build / JDK 17 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()
Build / JDK 17 and Scala 2.13 / kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()
```
--
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
[GitHub] [kafka] peterwanner commented on pull request #11855: MINOR: bring producer batch.size back
Posted by GitBox <gi...@apache.org>.
peterwanner commented on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1065832199
@showuon could you look at this review? or as what you say, it's reaching the end of the release
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11855: MINOR: fix default producer batch.size config
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#discussion_r825377329
##########
File path: core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala
##########
@@ -123,4 +163,37 @@ class ConsoleProducerTest {
assertEquals("console-producer",
producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG))
}
+
+ @Test
+ def testBatchSizeOverride(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeOverride)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals("456",
+ producerConfig.getString(ProducerConfig.BATCH_SIZE_CONFIG))
+ }
+
+ @Test
+ def testBatchSizeSet(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeSet)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals("123",
+ producerConfig.getString(ProducerConfig.BATCH_SIZE_CONFIG))
+ }
+
+ @Test
+ def testDefaultBatchSize(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeDefault)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals(16*1024,
+ producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG))
+ }
+
+ @Test
+ def testBatchSizeNotSet(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeNotSet)
Review comment:
I think this test can be renamed to `testBatchSizeNotSetButMaxPartitionMemoryBytesSet` to be clear.
And also the config variable name can be `batchSizeNotSetButMaxPartitionMemoryBytesSet`
##########
File path: core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala
##########
@@ -123,4 +163,37 @@ class ConsoleProducerTest {
assertEquals("console-producer",
producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG))
}
+
+ @Test
+ def testBatchSizeOverride(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeOverride)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals("456",
+ producerConfig.getString(ProducerConfig.BATCH_SIZE_CONFIG))
Review comment:
Test failed with
```
ava.lang.ClassCastException: class java.lang.Integer cannot be cast to class java.lang.String (java.lang.Integer and java.lang.String are in module java.base of loader 'bootstrap')
```
It should be using getInt, and expected value should also be int 456, not string "456", like this:
`assertEquals(456, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG))`
Same comments to below tests.
##########
File path: core/src/main/scala/kafka/tools/ConsoleProducer.scala
##########
@@ -110,6 +110,10 @@ object ConsoleProducer {
props, ProducerConfig.SEND_BUFFER_CONFIG, config.options, config.socketBufferSizeOpt)
CommandLineUtils.maybeMergeOptions(
props, ProducerConfig.BUFFER_MEMORY_CONFIG, config.options, config.maxMemoryBytesOpt)
+ CommandLineUtils.maybeMergeOptions(
Review comment:
We can add a comment above this line, to mention why we set batch.size with two different option values. Ex:
// We currently have 2 options to set the batch.size value. We'll deprecate/remove one of them in KIP-717.
##########
File path: core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala
##########
@@ -123,4 +163,37 @@ class ConsoleProducerTest {
assertEquals("console-producer",
producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG))
}
+
+ @Test
+ def testBatchSizeOverride(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeOverride)
Review comment:
Maybe rename to `testBatchSizeOverriddenByMaxPartitionMemoryBytesValue`
And the config variable name.. hmm... I think `batchSizeOverride` is OK. What do you think?
##########
File path: core/src/main/scala/kafka/tools/ConsoleProducer.scala
##########
@@ -138,6 +142,12 @@ object ConsoleProducer {
.withOptionalArg()
.describedAs("compression-codec")
.ofType(classOf[String])
+ val batchSizeOpt = parser.accepts("batch-size", "Number of messages to send in a single batch if they are not being sent synchronously. "+
+ "please note that this opt will be replaced if max-partition-memory-bytes is also set")
Review comment:
nit: this opt -> this **option**
##########
File path: core/src/test/scala/unit/kafka/tools/ConsoleProducerTest.scala
##########
@@ -123,4 +163,37 @@ class ConsoleProducerTest {
assertEquals("console-producer",
producerConfig.getString(ProducerConfig.CLIENT_ID_CONFIG))
}
+
+ @Test
+ def testBatchSizeOverride(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeOverride)
+ val producerConfig = new ProducerConfig(ConsoleProducer.producerProps(config))
+ assertEquals("456",
+ producerConfig.getString(ProducerConfig.BATCH_SIZE_CONFIG))
+ }
+
+ @Test
+ def testBatchSizeSet(): Unit = {
+ val config = new ConsoleProducer.ProducerConfig(batchSizeSet)
Review comment:
renamed to `testBatchSizeSetButMaxPartitionMemoryBytesNotSet`
And also the config variable name `batchSizeSet` -> `batchSizeSetButMaxPartitionMemoryBytesNotSet`
##########
File path: core/src/main/scala/kafka/tools/ConsoleProducer.scala
##########
@@ -260,6 +270,7 @@ object ConsoleProducer {
DefaultCompressionCodec.name
else compressionCodecOptionValue
else NoCompressionCodec.name
+ val batchSize = options.valueOf(batchSizeOpt)
Review comment:
This is not used. It can be removed.
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11855: MINOR: bring producer batch.size back
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#discussion_r825266382
##########
File path: core/src/main/scala/kafka/tools/ConsoleProducer.scala
##########
@@ -138,6 +140,12 @@ object ConsoleProducer {
.withOptionalArg()
.describedAs("compression-codec")
.ofType(classOf[String])
+ val batchSizeOpt = parser.accepts("batch-size", "Number of messages to send in a single batch if they are not being sent synchronously. "+
+ "please note that this opt will be replaced if max-partition-memory-bytes is also set")
+ .withRequiredArg
+ .describedAs("size")
+ .ofType(classOf[java.lang.Integer])
+ .defaultsTo(200)
Review comment:
We should make the default value the same the `max-partition-memory-bytes`, which is `16 * 1024`, so that we can do this:
`this opt will be replaced if max-partition-memory-bytes is also set`
##########
File path: core/src/main/scala/kafka/tools/ConsoleProducer.scala
##########
@@ -110,6 +110,8 @@ object ConsoleProducer {
props, ProducerConfig.SEND_BUFFER_CONFIG, config.options, config.socketBufferSizeOpt)
CommandLineUtils.maybeMergeOptions(
props, ProducerConfig.BUFFER_MEMORY_CONFIG, config.options, config.maxMemoryBytesOpt)
+ CommandLineUtils.maybeMergeOptions(
+ props, ProducerConfig.BATCH_SIZE_CONFIG, config.options, config.maxPartitionMemoryBytesOpt)
Review comment:
We can try to merge the `config.batchSizeOpt` first, and then merge `config.maxPartitionMemoryBytesOpt`. Like this:
```java
CommandLineUtils.maybeMergeOptions(
props, ProducerConfig.BATCH_SIZE_CONFIG, config.options, config.batchSizeOpt)
CommandLineUtils.maybeMergeOptions(
props, ProducerConfig.BATCH_SIZE_CONFIG, config.options, config.maxPartitionMemoryBytesOpt)
```
This way, the batch.size value will be replaced with `maxPartitionMemoryBytesOpt` option value (if set), otherwise, the `batchSizeOpt` option value will be used (if set), otherwise, default value will be used (i.e. 16 * 1024).
These following cases will have such results:
1. bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test **--batch-size 123**
--> batch.size = 123
2. bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test **--max-partition-memory-bytes 456**
--> batch.size = 456
3. bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test **--max-partition-memory-bytes 456 --batch-size 123**
--> batch.size = 456
4. bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
--> batch.size = 16 * 1024
Does that make sense?
If so, please help add some tests in `ConsoleProducerTest` for the above 4 cases.
Thank you.
--
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
[GitHub] [kafka] showuon commented on pull request #11855: MINOR: set `batch-size` option value into batch.size config in consoleProducer
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1066052188
Will wait for another day to merge to see if there are other people want to have another look. 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
[GitHub] [kafka] showuon commented on pull request #11855: MINOR: Clarify producer batch.size in upgrade docs
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1063601218
@peterwanner , since it's reaching the end of the release. Please let me know if you can't work on it before next weekend. Thank you.
--
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
[GitHub] [kafka] showuon commented on pull request #11855: MINOR: Clarify producer batch.size in upgrade docs
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1062600064
@peterwanner , since this `removing existing public option` needs KIP discussion/voting, and the v3.2.0 already KIP freezed, I think we should revert the change we did in https://github.com/apache/kafka/pull/11517. That means, we don't need the change in `docs/upgrade.html` now. Sorry for not reminding you during the previous PR!
So, in short, I need your help to submit a PR (or use this PR) to:
1. revert the change in https://github.com/apache/kafka/pull/11517 to bring the `batchSizeOpt` argument back
2. I found we accidentally remove the line [here](https://github.com/apache/kafka/pull/11517/files#diff-3b7393aaaffd466b3bfc53c8d95109a206442160b878bbdaa0b0332257afaebaL113-L114), where user can set the `max-partition-memory-bytes` to change the `batch.size` in producer. Anyway, please bring it back.
3. Fix the bug in `batchSizeOpt`, by feeding in to the producer `batch.size` config if `batch-size` option is set. So that user can set the `batch.size` via `batch-size` option or `max-partition-memory-bytes` option.
4. In `batch.size` option description, please note that this value in this value will be replaced if `max-partition-memory-bytes` is also set.
5. The result: user can still work well using `batch.size` option (if set), and now, the option will take effect in the producer. And most importantly, it won't break any existing script.
Does that make sense?
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
[GitHub] [kafka] showuon edited a comment on pull request #11855: MINOR: Clarify producer batch.size in upgrade docs
Posted by GitBox <gi...@apache.org>.
showuon edited a comment on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1062600064
@peterwanner , since this `removing existing public option` needs KIP discussion/voting, and the v3.2.0 already KIP freezed, I think we should revert the change we did in https://github.com/apache/kafka/pull/11517. That means, we don't need the change in `docs/upgrade.html` now. Sorry for not reminding you during the previous PR!
So, in short, I need your help to submit a PR (or use this PR) to:
1. revert the change in https://github.com/apache/kafka/pull/11517 to bring the `batchSizeOpt` argument back
2. I found we accidentally remove the line [here](https://github.com/apache/kafka/pull/11517/files#diff-3b7393aaaffd466b3bfc53c8d95109a206442160b878bbdaa0b0332257afaebaL113-L114), where user can set the `max-partition-memory-bytes` to change the `batch.size` in producer. Anyway, please bring it back.
3. Fix the bug in `batchSizeOpt`, by feeding in to the producer `batch.size` config if `batch-size` option is set. So that user can set the `batch.size` via `batch-size` option or `max-partition-memory-bytes` option.
4. In `batch.size` option description, please note that this value in this value will be replaced if `max-partition-memory-bytes` is also set.
5. The result: user can still work well using `batch.size` option (if set), and now, the option will take effect in the producer. And most importantly, it won't break any existing script.
6. For the default value in `batch-size`, we can ignore that. After all, we will (and should) deprecate it soon.
Does that make sense?
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
[GitHub] [kafka] peterwanner commented on pull request #11855: MINOR: Clarify producer batch.size in upgrade docs
Posted by GitBox <gi...@apache.org>.
peterwanner commented on pull request #11855:
URL: https://github.com/apache/kafka/pull/11855#issuecomment-1063660862
> @peterwanner , since it's reaching the end of the release. Please let me know if you can't work on it before next weekend. Thank you.
@showuon ok I will finish in the afternoon, please wait for a seconds. Thank you so much for your reminder
--
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
[GitHub] [kafka] showuon merged pull request #11855: MINOR: set `batch-size` option value into batch.size config in consoleProducer
Posted by GitBox <gi...@apache.org>.
showuon merged pull request #11855:
URL: https://github.com/apache/kafka/pull/11855
--
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