In Kafka Streams, configuration classes typically follow a fluent API
pattern to ensure a consistent and intuitive developer experience.
However, the current implementation of
`org.apache.kafka.streams.KafkaStreams$CloseOptions` deviates from this
convention by exposing a public constructor, breaking the uniformity
expected across the API.
To address this inconsistency, we propose introducing a new
`CloseOptions` class that adheres to the fluent API style, replacing the
existing implementation. The new class will retain the existing
`timeout(Duration)` and `leaveGroup(boolean)` methods but will enforce
fluent instantiation and configuration. Given the design shift, we will
not maintain backward compatibility with the current class.
This change aligns with the goal of standardizing configuration objects
across Kafka Streams, offering developers a more cohesive and
predictable API.
Reviewers: Bill Bejeck<bbejeck@apache.org>
With changes to the consumer protocol, rebalance may not necessarily
result in a "stop the world". Thus, the method for calculating pause
time in `ConsumerPerformance#ConsumerPerfRebListener` needs to be
modified.
Stop time is only recorded if `assignedPartitions` is empty.
Reviewers: Andrew Schofield <aschofield@confluent.io>
Improve `MetadataVersion.fromVersionString()` to take an
`enableUnstableFeature` flag, and enable `FeatureCommand` and
`StorageTool` to leverage the exception message from
`fromVersionString`.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
*What*
In the implementation of KIP-1147 for console tools -
https://github.com/apache/kafka/pull/20479/files#diff-85b87c675a4b933e8e0e05c654d35d60e9cfd36cebe3331af825191b2cc688ee,
we missed adding unit tests for verifying the new
"`--formatter-property`" option.
Thanks to @Yunyung for pointing this out.
PR adds unit tests to both `ConsoleConsumerOptionsTest` and
`ConsoleShareConsumerOptionsTest` to verify the same.
Reviewers: Jhen-Yung Hsu <jhenyunghsu@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
Small improvements to the argument descriptions in the usage messages
for the console producer/consumer tools.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
*What*
https://issues.apache.org/jira/browse/KAFKA-19623
- The PR implements KIP-1147
(https://cwiki.apache.org/confluence/display/KAFKA/KIP-1147%3A+Improve+consistency+of+command-line+arguments)
for the console tools i.e. `ConsoleProducer`, `ConsoleConsumer` and
`ConsoleShareConsumer`.
- Currently the previous names for the options are still usable but
there will be warning message stating those are deprecated and will be
removed in a future version.
- I have added unit tests and also manually verified using the console
tools that things are working as expected.
Reviewers: Andrew Schofield <aschofield@confluent.io>, Jhen-Yung Hsu
<jhenyunghsu@gmail.com>, Jimmy Wang
<48462172+JimmyWang6@users.noreply.github.com>
This is actually fixing a difference between the old and the new
assignor. Given the assignment ordering, the legacy assignor has a
preference for range-style assignments built in, that is, assigning
C1: 0_0, 1_0 C2: 0_1, 1_1
instead of
C1: 0_0, 0_1 C2: 1_0, 1_1
We add tests to both assignors to check for this behavior, and improve
the new assingor by enforcing corresponding orderings.
Reviewers: Bill Bejeck <bill@confluent.io>
# Description
`kafka-acls.sh` doesn't print message about duplicate authorization.
# Changes
Now the cli searches for existing AclBinding, prints duplicate bindings,
and removes them from the adding list.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
We add the three main changes in this PR
- Disallowing null values for most LIST-type configurations makes sense,
since users cannot explicitly set a configuration to null in a
properties file. Therefore, only configurations with a default value of
null should be allowed to accept null.
- Disallowing duplicate values is reasonable, as there are currently no
known configurations in Kafka that require specifying the same value
multiple times. Allowing duplicates is both rare in practice and
potentially confusing to users.
- Disallowing empty list, even though many configurations currently
accept them. In practice, setting an empty list for several of these
configurations can lead to server startup failures or unexpected
behavior. Therefore, enforcing non-empty lists helps prevent
misconfiguration and improves system robustness.
These changes may introduce some backward incompatibility, but this
trade-off is justified by the significant improvements in safety,
consistency, and overall user experience.
Additionally, we introduce two minor adjustments:
- Reclassify some STRING-type configurations as LIST-type, particularly
those using comma-separated values to represent multiple entries. This
change reflects the actual semantics used in Kafka.
- Update the default values for some configurations to better align with
other configs.
These changes will not introduce any compatibility issues.
Reviewers: Jun Rao <junrao@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
Should be `headers.separator=<headers.separator>` instead of
`headers.separator=<line.separator>`
Reviewers: Kuan-Po Tseng <brandboat@gmail.com>, Ken Huang
<s7133700@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
The `kafka-share-groups.sh` tool checks whether a topic already has a
start-offset in the share group when resetting offsets. This is not
necessary. By removing the check, it is possible to set a start offset
for a topic which has not yet but will be subscribed in the future, thus
initialising the consumption point.
There is still a small piece of outstanding work to do with resetting
the offset for a non-existent group which should also create the group.
A subsequent PR will be used to address that.
Reviewers: Jimmy Wang <48462172+JimmyWang6@users.noreply.github.com>,
Lan Ding <isDing_L@163.com>, Apoorv Mittal <apoorvmittal10@gmail.com>
Ensure that metrics are retrieved and displayed (when requested) before
ShareConsumer.close() is called. This is important because metrics are
technically supposed to be removed on ShareConsumer.close(), which means
retrieving them after close() would yield an empty map.
Related to https://github.com/apache/kafka/pull/20267.
Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>
#20390 Replace the -`-producer.config` for the verifiable producer and
`--consumer.config` option by `--command-config` for the verifiable
consumer. However, for e2e tests targeting older broker versions, the
original configuration should still be used.
Fix the following tests:
`consumer_protocol_migration_test.py`、`compatibility_test_new_broker_test.py`
and `upgrade_test.py`.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Manikumar Reddy <manikumar.reddy@gmail.com>
Refactor help and version handling in command-line tools by replacing
duplicate code with `CommandLineUtils#maybePrintHelpOrVersion`.
Reviewers: TengYao Chi <kitingiao@gmail.com>, Ken Huang
<s7133700@gmail.com>, Jhen-Yung Hsu <jhenyunghsu@gmail.com>, Chia-Ping
Tsai <chia7712@gmail.com>
jira: https://issues.apache.org/jira/browse/KAFKA-19011 kip:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1172%3A+Improve+EndToEndLatency+tool
This PR improves the usability and maintainability of the
`kafka-e2e-latency.sh` tool:
- Replaces fixed-index argument parsing with a proper argument parser
(joptsimple)
- Adds support for configuring:
- -record-key-size: size of the message key
- -num-headers: number of headers per message
- -record-header-key-size: size of each header key
- -record-header-size: size of each header value
- Renames existing arguments to align with Kafka CLI conventions:
- broker_list → bootstrap-server
- num_messages → num-records
- message_size_bytes → record-size
- properties_file → command-config
-
Reviewers: Jhen-Yung Hsu <jhenyunghsu@gmail.com>, Ken Huang
<s7133700@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
Ensure that metrics are retrieved and displayed (when requested) before
`Consumer.close()` is called. This is important because metrics are
technically supposed to be removed on `Consumer.close()`, which means
retrieving them _after_ `close()` would yield an empty map.
Reviewers: Andrew Schofield <aschofield@confluent.io>
issue: [KAFKA-19616](https://issues.apache.org/jira/browse/KAFKA-19616)
**why**: validate log compaction works correctly with compressed data.
**what**: adds compression config options to `LogCompactionTester` tool
and extends test coverage to validate log compaction with different
compression types and levels.
Reviewers: TengYao Chi <kitingiao@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
As described in
[jira](https://issues.apache.org/jira/browse/KAFKA-19625), this PR
implements replace `consumer.config` and `producer.config` with
`command-config` for kafka-verifiable-producer.sh and
kafka-verifiable-consumer.sh.
Reviewers: Andrew Schofield <aschofield@confluent.io>
This is the first part of the implementation of
[KIP-1023](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1023%3A+Follower+fetch+from+tiered+offset)
The purpose of this pull request is for the broker to start returning
the correct offset when it receives a -6 as a timestamp in a ListOffsets
API request.
Added unit tests for the new timestamp.
Reviewers: Kamal Chandraprakash <kamal.chandraprakash@gmail.com>
This PR applies the same partition leader check for `StreamsGroupCommand` as
`ShareGroupCommand` and `ConsumerGroupCommand` to avoid the command
execution timeout.
Reviewers: Lucas Brutschy <lucasbru@apache.org>
This patch adds the include argument to ConsumerPerformance tool.
ConsoleConsumer and ConsumerPerformance serve different purposes but
share common functionality for message consumption. Currently, there's
an inconsistency in their command-line interfaces:
- ConsoleConsumer supports an --include argument that allows users to
specify a regular expression pattern to filter topics for consumption
- ConsumerPerformance lacks this topic filtering capability, requiring
users to specify a single topic explicitly via --topic argument
This inconsistency creates two problems:
- Similar tools should provide similar topic selection capabilities for
better user experience
- Users cannot test consumer performance across multiple topics or
dynamically matching topic sets, making it difficult to test realistic
scenarios
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
issue: https://github.com/apache/kafka/pull/19905#discussion_r2282202312
What: Change `String[] topics` to `Set<String> topics` throughout
`LogCompactionTester`. Why: `Set<String>` is more modern and reduces the
need for array-to-collection conversions.
Reviewers: Ken Huang <s7133700@gmail.com>, TengYao Chi
<kitingiao@gmail.com>, Jhen-Yung Hsu <jhenyunghsu@gmail.com>, Lan Ding
<isDing_L@163.com>, Kuan-Po Tseng <brandboat@gmail.com>, Chia-Ping
Tsai <chia7712@gmail.com>
The ReassignPartitionsCommand shows the topic replicas on each broker.
When using the --generate command, it returns the current partition
replica assignment. However, the log directory for each current replica
is always shown as any. This makes it impossible for users to determine
which specific log directory is being used by each replica. Therefore,
we should fix this behavior.
```
Current partition replica assignment
{
"version": 1,
"partitions": [
{
"topic": "test1",
"partition": 0,
"replicas": [
4,
2
],
"log_dirs": [
"any",
"any"
]
}
]
}
```
This PR
```
Current partition replica assignment
{
"version": 1,
"partitions": [
{
"topic": "test1",
"partition": 0,
"replicas": [
4,
2
],
"log_dirs": [
"/tmp/kraft-broker-logs234",
"/tmp/kraft-broker-logs"
]
}
]
}
```
Reviewers: PoAn Yang <payang@apache.org>, Jhen-Yung Hsu
<jhenyunghsu@gmail.com>, TaiJuWu <tjwu1217@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
This PR aims at cleaning up the tools module further by getting rid of
some extra code which can be replaced by `record`
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
Refactor metric gauges instantiation to use lambda expressions instead
of ImmutableValue.
Reviewers: Ken Huang <s7133700@gmail.com>, TengYao Chi
<kitingiao@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
This PR do following:
1. Use correct json format to test.
2. make `PartitionReassignmentState` and `VerifyAssignmentResult.java`
become record
Reviewers: TengYao Chi <frankvicky@apache.org>, Ken Huang
<s7133700@gmail.com>, PoAn Yang <payang@apache.org>
Now that Kafka support Java 17, this PR makes some changes in tools
module. The changes in this PR are limited to only some files. A future
PR(s) shall follow. The changes mostly include:
- Collections.emptyList(), Collections.singletonList() and
Arrays.asList() are replaced with List.of()
- Collections.emptyMap() and Collections.singletonMap() are replaced
with Map.of()
- Collections.singleton() is replaced with Set.of()
Some minor changes to use the enhanced switch.
Sub modules targeted: tools/src/test
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>
See: https://github.com/apache/kafka/pull/20168#discussion_r2227310093
add follow test case:
Given a topic with three partitions, where partition `t-2` is offline,
if partitionsToReset contains only `t-1`, the method
filterNoneLeaderPartitions incorrectly includes `t-2` in the result,
leading to a failure in the tool.
Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Jhen-Yung Hsu
<jhenyunghsu@gmail.com>, Ken Huang <s7133700@gmail.com>, Andrew
Schofield <aschofield@confluent.io>
### Problem
The connect-plugin-path tool crashes with
`UnsupportedOperationException` when processing plugins that have
loadable classes but no ServiceLoader manifest files.
### Root Cause
Line 326 attempts to remove from an immutable `Collections.emptySet()`:
```java
nonLoadableManifests.getOrDefault(pluginDesc.className(),
Collections.emptySet()).remove(pluginDesc.type());
```
### Solution
Replace `Collections.emptySet()` with `new HashSet<>()` to provide a
mutable set.
Reviewers: Ken Huang <s7133700@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
[KAFKA-16717](https://issues.apache.org/jira/browse/KAFKA-16717) aims to
finish the AlterShareGroupOffsets for ShareGroupCommand part.
Reviewers: Lan Ding <isDing_L@163.com>, Chia-Ping Tsai
<chia7712@gmail.com>, TaiJuWu <tjwu1217@gmail.com>, Andrew Schofield
<aschofield@confluent.io>
In order to better analyze steady-state performance of Kafka, this PR
enables a warmup in the Producer Performance test. The warmup duration
is specified as a number of records that are a subset of the total
numRecords. If warmup records is greater than 0, the warmup is
represented by a second Stats object which holds warmup results. Once
warmup records have been exhausted, the test switches to using the
existing Stats object. At end of test, if warmup was enabled, the
summary of the whole test (warump + steady state) is printed followed by
the summary of the steady-state portion of the test. If no warmup is
used, summary prints don't change from existing behavior. This
contribution is an original work and is licensed to the Kafka project
under the Apache license
Testing strategy comprises new Java unit tests added to
ProducerPerformanceTests.java.
Reviewers: Kirk True <kirk@kirktrue.pro>, Federico Valeri
<fedevaleri@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
### Problem
The
`ShareGroupCommandTest.testDeleteShareGroupOffsetsArgsWithoutTopic()`,
`ShareGroupCommandTest.testDeleteShareGroupOffsetsArgsWithoutGroup()`,
`ResetStreamsGroupOffsetTest.testResetOffsetsWithoutGroupOption()`,
`DeleteStreamsGroupTest.testDeleteWithoutGroupOption()`,
`DescribeStreamsGroupTest.testDescribeWithoutGroupOption()` tests were
flaky due to a dependency on Set iteration order in error message
generation.
### Root Cause
The cleanup [commit](https://github.com/apache/kafka/pull/20091) that
replaced `new HashSet<>(Arrays.asList(...))` with `Set.of(...)` in
ShareGroupCommandOptions and StreamsGroupCommandOptions changed the
iteration characteristics of collections used for error message
generation:
This produces different orders like `[topic], [group]` vs `[group],
[topic]`, but the tests expected a specific order, causing intermittent
failures.
### Solution
Fix the root cause by ensuring deterministic error message generation
through alphabetical sorting of option names.
Reviewers: ShivsundarR <shr@confluent.io>, Ken Huang
<s7133700@gmail.com>, TengYao Chi <frankvicky@apache.org>
The descrption "REQUIRED: The number of messages to send or consume" is
not correct, since those tools do NOT send any records.
Reviewers: TengYao Chi <frankvicky@apache.org>
Changes: Rename `waitForTopic` to `waitTopicCreation` for better clarity
Reasons: To align with `waitTopicDeletion` Reference:
https://github.com/apache/kafka/pull/20108/files#r2221659660
Reviewers: Ken Huang <s7133700@gmail.com>, TengYao Chi
<frankvicky@apache.org>
Now that Kafka support Java 17, this PR makes some changes in tools
module. The changes in this PR are limited to only some files. A future
PR(s) shall follow.
The changes mostly include:
- Collections.emptyList(), Collections.singletonList() and
Arrays.asList() are replaced with List.of()
- Collections.emptyMap() and Collections.singletonMap() are replaced
with Map.of()
- Collections.singleton() is replaced with Set.of()
Sub modules targeted: tools/src/main
Reviewers: Ken Huang <s7133700@gmail.com>, Jhen-Yung Hsu
<jhenyunghsu@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
1. Add check leader missing logic in method
`ConsumerGroupCommand.ConsumerGroupService#prepareOffsetsToReset` in
order to fail quickly
2. Add some tests
Reviewers: TaiJuWu <tjwu1217@gmail.com>, Lan Ding <isDing_L@163.com>,
Ken Huang <s7133700@gmail.com>, Andrew Schofield
<aschofield@confluent.io>
For the Kafka Stream group commands, if delete topic requests fail due
to version mismatch, user will have to remove the topics manually by
first retrieving the relevant internal topics.
To assist the user, the internal topic names are now included as part of
the error message, so that the user could delete the internal topics
associated with this application directly.
Reviewers: TengYao Chi <frankvicky@apache.org>, Alieh Saeedi
<asaeedi@confluent.io>
1. Optimize the corresponding logic in the `ConsumerGroupCommand` by
first checking if a leader exists for the partition before invoking the
`admin.listOffsets`. Finally, concatenate the data and return
2. Add integration test, create a cluster with 3 brokers, then shutdown
a broker and observe whether the output meets the expectations
Reviewers: Ken Huang <s7133700@gmail.com>, PoAn Yang
<payang@apache.org>, TaiJuWu <tjwu1217@gmail.com>, Chia-Ping Tsai
<chia7712@gmail.com>
This fixes the flaky
`DescribeStreamsGroupTest.testDescribeMultipleStreamsGroupWithMembersAndVerboseOptions()`,
which sometimes fails due to `ERROR stream-thread Missing source topics:
Source topics customInputTopic2 are missing`
Reviewers: Bill Bejeck <bbejeck@apache.org>
"the action will only execute" is incorrect, as the admin still sends
the request. The "if-not-exists" flag is actually used to swallow the
exception
Reviewers: TengYao Chi <frankvicky@apache.org>, Nick Guo
<lansg0504@gmail.com>, Ken Huang <s7133700@gmail.com>
## Changes
- The partitions == 0 branch has been moved from **waitForTopic** to
**waitTopicDeletion**.
## Reasons
- Clarify the responsibility of each helper method makes the test code
easier to reason by moving the partitions == 0 logic from
**waitForTopic** into a dedicated method **waitTopicDeletion**.
Reviewers: Jhen-Yung Hsu <jhenyunghsu@gmail.com>, TaiJuWu
<tjwu1217@gmail.com>, TengYao Chi <kitingiao@gmail.com>, Ken Huang
<s7133700@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
### About
Within `ShareConsumerPerformance.java`, all the share consumers run with
within an executorService object and when we
perform `executorService.submit()`, we do not store this future and
exception would be recovered only when we do a future.get() in this
case. I believe this is a shortcoming
in `ShareConsumerPerformance.java` which needs to be improved.
Reviewers: Andrew Schofield <aschofield@confluent.io>
Ken Huang
Lan Ding
Shivsundar R
ally heev
Andrew Schofield
Lucas Brutschy
Chang-Yu Huang
Levani Kokhreidze
jimmy
Hong-Yi Chen
Nick Guo
Kirk True
Yunchi Pang
Jhen-Yung Hsu
jimmy
Abhijeet Kumar
Federico Valeri
Sanskar Jhajharia
Ming-Yen Chung
TaiJuWu
Tsung-Han Ho (Miles Ho)
majialong
Maros Orsak
Matt Welch
Chang-Chi Hsu
xijiu
lucliu1108
Alieh Saeedi
Xuan-Zhang Gong
Abhinav Dixit