Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Add the following methods to the org.apache.kafka.client.admin.AdminClient  interface, mostly backed by the same implementations as the consumer group API.

Method signature

Description

RPC used

AlterStreamsGroupOffsetsResult alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long> offsets)Alter offset information for a streams group.OffsetCommit
AlterStreamsGroupOffsetsResult alterStreamsGroupOffsets(String groupId, Map<TopicPartition, Long> offsets, AlterStreamsGroupOffsetsOptions options) Alter offset information for a streams group.OffsetCommit
DeleteStreamsGroupOffsetsResult deleteStreamsGroupOffsets(String groupId, Set<String> topics)Delete offset information for a set of topics in a streams group.OffsetDelete
DeleteStreamsGroupOffsetsResult deleteStreamsGroupOffsets(String groupId, Set<String> topics, DeleteStreamsGroupOffsetsOptions options) Delete offset information for a set of topics in a streams group.OffsetDelete
DeleteStreamsGroupResult deleteStreamsGroups(Collection<String> groupIds)Delete streams groups from the cluster.DeleteGroups
DeleteStreamsGroupResult deleteStreamsGroups(Collection<String> groupIds, DeleteStreamsGroupOptions options) Delete streams groups from the cluster.DeleteGroups
DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds)Describe some streams groups in the cluster.StreamsGroupDescribe
DescribeStreamsGroupsResult describeStreamsGroups(Collection<String> groupIds, DescribeStreamsGroupsOptions options) Describe some streams groups in the cluster.StreamsGroupDescribe
ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs)List the streams group offsets available in the cluster for the specified Streams groups.OffsetFetch
ListStreamsGroupOffsetsResult listStreamsGroupOffsets(Map<String, ListStreamsGroupOffsetsSpec> groupSpecs, ListStreamsGroupOffsetsOptions options) List the streams group offsets available in the cluster for the specified Streams groups.OffsetFetch
ListStreamsGroupsResult listStreamsGroups()List the streams groups available in the cluster.ListGroups
ListStreamsGroupsResult listStreamsGroups(ListStreamsGroupsOptions options) List the streams groups available in the cluster.ListGroups

These correspond to existing APIs for consumer groups and some of the code will be shared. The main reason for duplicating them is the naming, as the current API uses "consumerGroup" in the name, despite the RPCs just using "group". There are some differences:

...

Another case is added to the org.apache.kafka.common.GroupType  enum:

Enum constant

Description

Streams("Streams") Streams group

StreamsGroupState

A new enum org.apache.kafka.common.StreamsGroupState  is added:

Enum constant

DEAD 

EMPTY 

STABLE 

INITIALIZING

ASSIGNING

RECONCILING

UNKNOWN 

Exceptions

The following new exceptions are added to the org.apache.kafka.common.errors package, corresponding to the new error codes in the Kafka protocol.

...

Option

Description

--version

Display Kafka version.

--all-input-topics

Reset all input topic in the reset offset processUse with --reset-offsets or --delete-offsets. If specified, includes all input topics of the streams group, as stored in by the topology metadata on the broker.

--input-topics <String: topics>

The input topic that should be included in the reset offset process.Use with --reset-offsets or --delete-offsets. Comma-separated list of user input topics. For these topics, the tool by default will reset the offset to the earliest available offset, or delete the offsets. Reset to other offset position by appending other reset offset option, ex: --input-topics foo --shift-by 5. 

--internal-topics <String: topics>

The internal topics that should be deleted. By default, include all internal topics of the streams group, as stored by the topology metadata on the broker.

--topic <String: topic>

The topic whose streams group information should be deleted or included in the reset offset process. In `reset-offsets` case, partitions can be specified using this format: `topic1:0,1,2`, where 0,1,2 are the partitions to be included.Use with --delete. Comma-separated list of internal topics to delete. Must be a subset of the internal topics marked for deletion by the default behaviour (do a dry-run without this option to view these topics). 

--to-offset <Long: offset>

Reset input topic offsets to a specific offset

--to-latest

Reset input topic offsets to latest offset.

--to-earliest

Reset input topic offsets to earliest offset.

--by-duration <String: duration>

Reset input topic offsets to offset by duration from current timestamp. Format: 'PnDTnHnMnS'

--to-datetime <String: datetime>

Reset input topic offsets to offset from datetime. Format: 'YYYY-MM-DDTHH:mm:SS.sss'.

--from-file

Reset input topic offsets to values defined in CSV file.

--shift-by <Long: n>

Reset input topic offsets shifting current offset by 'n', where 'n' can be positive or negative.

--timeout <Long: timeout (ms)>

The timeout that can be set for some use cases. For example, it can be used when describing the group to specify the maximum amount of time in milliseconds to wait before the group stabilizes (when the group is just created, or is going through some changes). (default: 5000)   

--state [String]

When specified with '--describe', includes the state of the group. When specified with '--list', it displays the state of all groups. It can also be used to list groups with specific states. The valid values are 'Empty', 'Initializing', 'Reconciling', 'Assigning', 'Stable' and 'Dead'.

--reset-offsets

Reset input topic offsets of streams group. Supports one streams group at a time, and instances should be inactive.

You must choose one of the following reset specifications: --to-datetime, --by-duration, --to-earliest, --to-latest, --shift-by, --from-file, --to-current, --to-offset.

To define the scope use --all-input-topics or --input-topics. One scope must be specified unless you use '--from-file'.

Fails if neither '--dry-run' nor '–execute' is specified.

--offsets

Describe the group and list all input topic partitions in the group along with their offset lag. This is the default sub-action of --describe and may be used with the '--describe' option only.

--members

Describe members of the group. This option may be used with the '--describe' option only.

--list

List all streams groups.

--help

Print usage information.

--group <String: streams group ID>

The group ID (application ID) of the streams ID we wish to act on.

--execute

Execute operation. Supported operations: reset-offsets.

--dry-run

Only show results without executing changes on streams groups. Supported operations: reset-offsets.

--describe

Describe streams group and list offset lag (number of records not yet processed) related to given group.

--delete-offsets

Delete offsets of streams group. Supports one streams group at the time.

To define the scope use --all-input-topics or --input-topics. One scope must be specified unless you use '--from-file'.

--delete

Pass in a group to delete entire streams group. For instance --group g1. Deletes offsets, internal topics of the streams group and and topology and ownership information on the broker. Instances should be inactive.

--command-config <String: command config property file>

Property file containing configs to be passed to Admin Client.

--bootstrap-server <String: server to connect to>

REQUIRED: The server(s) to connect to.

...

$ kafka-streams-groups.sh --bootstrap-server localhost:9092 --list

To delete the inactive streams group S1 , which deletes all offsets, internal topics, assignment and topology metadata on the broker:

$ kafka-streams-groups.sh --bootstrap-server localhost:9092 --group S1 --delete

To delete the offsets from input topics of information for topic T1  from inactive streams group S1 , which essentially resets the consumption of this topic in the streams group:

$ kafka-streams-groups.sh --bootstrap-server localhost:9092 --group S1 --delete-offsets --all-input-topics

To set the starting offset for consuming topic T1  in inactive the input topics of inactive streams group S1  to a specific date and time:

$ kafka-streams-groups.sh --bootstrap-server localhost:9092 --group S1 --reset-offsets --all-input-topics --to-datetime datetime 1999-12-31T23:57:00.000 --execute

...