You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "CCCLL (via GitHub)" <gi...@apache.org> on 2026/10/14 06:17:01 UTC
[PR] MINOR: Stop console producer from overriding configured compression.type [kafka]
CCCLL opened a new pull request, #23813:
URL: https://github.com/apache/kafka/pull/23813
`kafka-console-producer.sh` always sets `compression.type` from its
`--compression-codec` option, which falls back to `none` when the
option is not given. A `compression.type` set with `--command-property`
or in the `--command-config` file (or with the deprecated
`--producer-property` and `--producer.config`) is therefore silently
replaced and the records are sent uncompressed. For example,
`kafka-dump-log.sh` shows `compresscodec: none` for the records
produced by
kafka-console-producer.sh --bootstrap-server localhost:9092 \
--topic test --command-property compression.type=zstd
KAFKA-6627 fixed the same problem for the other producer options of
the tool, which go through `CommandLineUtils.maybeMergeOptions`, but
`--compression-codec` was left out. It does not fit that helper as is,
because the option means `gzip` when it is given without a value.
Apply the same precedence here: `--compression-codec` wins when it is
given, otherwise a configured `compression.type` is kept, and `none`
is used only when neither is set.
The new tests cover the default, both ways of configuring the
compression type and the precedence of `--compression-codec`. The
tests for `--command-property` and `--command-config` fail without the
fix.
--
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] MINOR: Stop console producer from overriding configured compression.type [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #23813:
URL: https://github.com/apache/kafka/pull/23813#discussion_r4275432259
##########
tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java:
##########
@@ -388,6 +388,59 @@ public void testCommandPropertyOverridesConfig() throws IOException {
assertEquals(16384, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
}
+ @Test
+ public void testDefaultCompressionType() throws IOException {
Review Comment:
I think this test never fails because the default config is always none ?
--
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] MINOR: Stop console producer from overriding configured compression.type [kafka]
Posted by "CCCLL (via GitHub)" <gi...@apache.org>.
CCCLL commented on code in PR #23813:
URL: https://github.com/apache/kafka/pull/23813#discussion_r4278658407
##########
tools/src/main/java/org/apache/kafka/tools/ConsoleProducer.java:
##########
@@ -392,7 +392,9 @@ Properties producerProps() throws IOException {
props.putAll(parseKeyValueArgs(options.valuesOf(commandPropertyOpt)));
props.put(BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOpt));
- props.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
+ if (options.has(compressionCodecOpt) || !props.containsKey(COMPRESSION_TYPE_CONFIG)) {
+ props.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
Review Comment:
Before this change such a value was silently replaced by `none`, so the tool started and sent the records uncompressed. Now it reaches `ProducerConfig` like any other producer property, and the tool exits with a `ConfigException` before sending anything:
```
org.apache.kafka.common.config.ConfigException: Invalid value foo for configuration compression.type: String must be one of: none, gzip, snappy, lz4, zstd
```
An empty value (`--command-property compression.type=` or `compression.type=` in the config file) fails the same way. This is what already happens for `--compression-codec foo` or `--command-property acks=foo`. I added `testInvalidCompressionTypeFromCommandProperty` for it.
##########
tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java:
##########
@@ -388,6 +388,59 @@ public void testCommandPropertyOverridesConfig() throws IOException {
assertEquals(16384, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
}
+ @Test
+ public void testDefaultCompressionType() throws IOException {
Review Comment:
Right, `none` is also the producer's default, so that test could not catch a regression here. I removed it.
##########
tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java:
##########
@@ -388,6 +388,59 @@ public void testCommandPropertyOverridesConfig() throws IOException {
assertEquals(16384, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
}
+ @Test
+ public void testDefaultCompressionType() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BOOTSTRAP_SERVER_VALID_ARGS);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("none", producerConfig.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ }
+
+ @Test
+ public void testCompressionTypeFromCommandProperty() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "compression.type=zstd"
+ };
+
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("zstd", producerConfig.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ }
+
+ @Test
+ public void testCompressionTypeFromCommandConfig() throws IOException {
+ File propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("compression.type", "lz4"));
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("lz4", producerConfig.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ }
+
+ @Test
+ public void testCompressionCodecOverridesCommandProperty() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "compression.type=zstd",
+ "--compression-codec", "lz4"
+ };
+
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("lz4", producerConfig.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ }
+
Review Comment:
Added tests for a bare `--compression-codec` (it means `gzip` and still takes precedence over `compression.type` from `--command-property`) and for the deprecated `--producer-property` and `--producer.config`.
--
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] MINOR: Stop console producer from overriding configured compression.type [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #23813:
URL: https://github.com/apache/kafka/pull/23813#discussion_r4275420857
##########
tools/src/main/java/org/apache/kafka/tools/ConsoleProducer.java:
##########
@@ -392,7 +392,9 @@ Properties producerProps() throws IOException {
props.putAll(parseKeyValueArgs(options.valuesOf(commandPropertyOpt)));
props.put(BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOpt));
- props.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
+ if (options.has(compressionCodecOpt) || !props.containsKey(COMPRESSION_TYPE_CONFIG)) {
+ props.put(COMPRESSION_TYPE_CONFIG, compressionCodec());
Review Comment:
How would be the behaviour incase of empty or invalid compression.type ?
--
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] MINOR: Stop console producer from overriding configured compression.type [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #23813:
URL: https://github.com/apache/kafka/pull/23813#discussion_r4275444781
##########
tools/src/test/java/org/apache/kafka/tools/ConsoleProducerTest.java:
##########
@@ -388,6 +388,59 @@ public void testCommandPropertyOverridesConfig() throws IOException {
assertEquals(16384, producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
}
+ @Test
+ public void testDefaultCompressionType() throws IOException {
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(BOOTSTRAP_SERVER_VALID_ARGS);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("none", producerConfig.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ }
+
+ @Test
+ public void testCompressionTypeFromCommandProperty() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "compression.type=zstd"
+ };
+
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("zstd", producerConfig.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ }
+
+ @Test
+ public void testCompressionTypeFromCommandConfig() throws IOException {
+ File propsFile = ToolsTestUtils.tempPropertiesFile(Map.of("compression.type", "lz4"));
+
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-config", propsFile.getAbsolutePath()
+ };
+
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("lz4", producerConfig.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ }
+
+ @Test
+ public void testCompressionCodecOverridesCommandProperty() throws IOException {
+ String[] args = new String[]{
+ "--bootstrap-server", "localhost:9092",
+ "--topic", "test",
+ "--command-property", "compression.type=zstd",
+ "--compression-codec", "lz4"
+ };
+
+ ConsoleProducerOptions opts = new ConsoleProducerOptions(args);
+ ProducerConfig producerConfig = new ProducerConfig(opts.producerProps());
+
+ assertEquals("lz4", producerConfig.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ }
+
Review Comment:
In general more tests are needed for ex a bare --compression-codec (no value).
and the deprecated --producer-property and --producer.config paths
--
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