DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
Status
Current state: Under discussion
Discussion thread: here
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Apache Kafka command-line tools provide the ability to reset offsets for consumer groups, share groups and streams groups. However, the usability of these tools is not great, particularly when offset information is exported to a file and then imported again.
For example, consider how the user exports offset information from one consumer group and resets another consumer group using those offsets. The commands look like this:
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group CG1 --export --topic T1 --to-current --dry-run > offsets.csv $ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group CG2 --from-file offsets.csv --topic T1 --execute
The first command exports the offset information for consumer group CG1 and topic T1, at the current position, in CSV format and then redirects it to a file called "offsets.csv" . The second command reads the offset information from that file and executes the reset operation on the offsets for consumer group CG2 and topic T1.
The first command is the strange one. Given that the command is really just exporting offset information, it's not resetting offsets at all. But that's really just because --dry-run was specified. If --execute had been specified, the command WOULD have reset the offsets on consumer group CG1, as well as exporting the offset information. The --export option doesn't export at all. What it does is format the output in CSV format rather than a human-readable table.
There is another usability problem which is that the dry-run operation for resetting offsets cannot easily check the group type. Resetting offsets can be used on a non-existent group as a way of initialising the group's position. If the user performs the dry-run reset operation for a share group when that group ID is actually a different type of group, the command works, but when they actually execute the reset operation, it fails because the group type is wrong. Really, we'd like to catch the incorrect group type in both cases, and to do this efficiently needs a change in the Admin API.
Proposed Changes
This KIP proposes introducing a new --export-offsets option on bin/kafka-consumer-groups.sh , bin/kafka-share-groups.sh and bin/kafka-streams-groups.sh. This new option does not reset the offsets of the group, it just captures the offset information and exports it in CSV form. There is no need to specify --dry-run nor --execute because they are not relevant to the export operation. The --export suboption of --reset-offsets is deprecated for removal in Apache Kafka 5.0.
The example then becomes:
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --export-offsets --group CG1 --topic T1 --to-current > offsets.csv $ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group CG2 --from-file offsets.csv --topic T1 --execute
The output format is unchanged. The output from --export-offsets is exactly the same as --reset-offsets --export generates.
In order to improve the usability of the tools further, a new admin client method called describeGroups is added which returns group descriptions for a set of groups, regardless of group type. This makes it easy to check whether a group exists and discover the group type, something that is not easy to achieve with the admin client today.
Public Interfaces
Client API changes
Admin client
Add the following methods on the org.apache.kafka.clients.admin.Admin interface.
| Method signature | Description |
|---|---|
DescribeGroupsResult describeGroups(Collection<String> groupIds) | Describe some groups in the cluster. |
DescribeGroupsResult describeGroups(Collection<String> groupIds, DescribeGroupsOptions options) | Describe some groups in the cluster. |
This method will only be supported when talking to a broker which supports the new DescribeGroupsGeneric RPC. This means that the command-line tools will need to handle the situation where the method throws UnsupportedVersionException and then fall back to the type-specific Admin.describeXXXGroups(groupIds) method instead.
Here are the method signatures.
/**
* Describe some groups in the cluster, with the default options.
*
* <p>This is a convenience method for {@link #describeGroups(Collection, DescribeGroupsOptions)}
* with default options. See the overload for more details.
*
* @param groupIds The IDs of the groups to describe.
* @return The DescribeGroupsResult.
*/
default DescribeGroupsResult describeGroups(Collection<String> groupIds) {
return describeGroups(groupIds, new DescribeGroupsOptions());
}
/**
* Describe some groups in the cluster.
*
* @param groupIds The IDs of the groups to describe.
* @param options The options to use when describing the groups.
* @return The DescribeGroupsResult.
*/
DescribeGroupsResult describeGroups(Collection<String> groupIds, DescribeGroupsOptions options);
DescribeGroupsOptions
package org.apache.kafka.clients.admin;
/**
* Options for the {@link Admin#describeGroups(Collection<String>, DescribeGroupsOptions)} call.
*/
public class DescribeGroupsOptions extends AbstractOptions<DescribeGroupsOptions> {
private boolean includeAuthorizedOperations;
public DescribeGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations) {
this.includeAuthorizedOperations = includeAuthorizedOperations;
return this;
}
public boolean includeAuthorizedOperations() {
return includeAuthorizedOperations;
}
}
DescribeGroupsResult
package org.apache.kafka.clients.admin;
/**
* The result of the {@link Admin#describeGroups(Collection<String>, DescribeGroupsOptions)} call.
*/
public class DescribeGroupsResult {
/**
* Return a future which yields all GroupDescription objects, if all the describes succeed.
*/
public KafkaFuture<Map<String, GroupDescription>> all() {
}
/**
* Return a map from group id to futures which yield group descriptions.
*/
public Map<String, KafkaFuture<GroupDescription>> describedGroups() {
}
}
GroupDescription
package org.apache.kafka.clients.admin;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.GroupState;
import org.apache.kafka.common.GroupType;
import org.apache.kafka.common.acl.AclOperation;
/**
* A detailed description of a single group in the cluster.
*/
public class GroupDescription {
public GroupDescription(String groupId, GroupType type, GroupState groupState, Node coordinator, Set<AclOperation> authorizedOperations);
/**
* The id of the group.
*/
public String groupId();
/**
* The group type (or the protocol) of this consumer group. It defaults
* to Classic if not provided by the server.
*/
public GroupType type();
/**
* The group state, or UNKNOWN if the state cannot be parsed.
*/
public GroupState groupState();
/**
* The group coordinator, or null if the coordinator is not known.
*/
public Node coordinator();
/**
* The authorized operations for this group, or null if that information is not known.
*/
public Set<AclOperation> authorizedOperations();
}
Command-line tools
Note that --export-offsets does not support --from-file as an offset specification, which differs from --reset-offsets . Of course, exporting offsets which have just been imported from a file would not make sense, which explains the difference.
kafka-consumer-groups.sh
The following options are added or modified by this KIP.
Option Description
------ -----------
--all-topics Consider all topics assigned to a
group in the `reset-offsets` or
`export-offsets` process.
--by-duration <String: duration> Reset or export offsets to offset by duration
from current timestamp. Format:
'PnDTnHnMnS'
--export Export operation execution to a CSV
file. Supported operations: reset-
offsets. (DEPRECATED: Use --export-offsets
instead.)
--export-offsets Export offset information in CSV format.
Supports one consumer group at a time.
You must choose one of the following
offset specifications: --to-datetime,
--by-duration, --to-earliest, --to-
latest, --shift-by, --to-current,
--to-offset.
--reset-offsets Reset offsets of consumer group.
Supports one consumer group at the
time, and instances should be
inactive
Has 2 execution options: --dry-run
(the default) to plan which offsets
to reset, and --execute to update
the offsets. Additionally, the --
export option (DEPRECATED) is used
to export the
results to a CSV format.
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-topics
or --topic. One scope must be
specified unless you use '--from-
file'.
--shift-by <Long: number-of-offsets> Reset or export offsets shifting current offset
by 'n', where 'n' can be positive or
negative.
--to-current Reset or export offsets to current offset.
--to-datetime <String: datetime> Reset or export offsets to offset from datetime.
Format: 'YYYY-MM-DDThh:mm:ss.sss'
--to-earliest Reset or export offsets to earliest offset.
--to-latest Reset or export offsets to latest offset.
--to-offset <Long: offset> Reset or export offsets to a specific offset.
kafka-share-groups.sh
The following options are added or modified by this KIP.
Note that the --by-duration and --shift-by options are also added to bring this tool in line with the other group tools.
Option Description
------ -----------
--all-topics Consider all topics assigned to a
share group in the 'reset-offsets'
or 'export-offsets' process.
--by-duration <String: duration> Reset or export offsets to offset by duration
from current timestamp. Format:
'PnDTnHnMnS'
--export Export operation execution to a CSV
file. Supported operations: reset-
offsets. (DEPRECATED: Use --export-offsets
instead.)
--export-offsets Export offsets of share group. Supports
one share group at the time.
You must choose one of the following
offset specifications: --to-current,
--to-datetime, --to-offset,
--to-earliest, --to-latest, --by-duration,
--shift-by.
To define the scope use --all-topics
or --topic.
--reset-offsets Reset offsets of share group. Supports
one share group at the time, and
instances must be inactive.
Has 2 execution options: --dry-run to
plan which offsets to reset, and --
execute to reset the offsets.
You must choose one of the following
reset specifications: --to-datetime,
--to-earliest, --to-latest, --to-offset,
--to-current, --by-duration, --shift-by.
To define the scope use --all-topics
or --topic.
Fails if neither '--dry-run' nor '--
execute' is specified.
--shift-by <Long: number-of-offsets> Reset or export offsets shifting current offset
by 'n', where 'n' can be positive or
negative.
--to-current Reset or export offsets to current offset.
--to-datetime <String: datetime> Reset or export offsets to offset from datetime.
Format: 'YYYY-MM-DDThh:mm:ss.sss'
--to-earliest Reset or export offsets to earliest offset.
--to-latest Reset or export offsets to latest offset.
--to-offset <Long: offset> Reset or export offsets to a specific offset.
--topic <String: topic> The topic whose offset information
should be deleted or included in the
reset or export offset process. When resetting
or exporting offsets, partitions can be specified
using this format: 'topic1:0,1,2',
where 0,1,2 are the partitions to be
included.
kafka-streams-groups.sh
Option Description
------ -----------
--all-input-topics Consider all source topics used in the
topology of the group. Supported
operations: delete-offsets, reset-
offsets, export-offsets.
--by-duration <String: duration> Reset or export offsets to offset by duration
from current timestamp. Format:
'PnDTnHnMnS'
--export Export operation execution to a CSV
file. Supported operations: reset-
offsets. (DEPRECATED: Use --export-offsets
instead.)
--export-offsets Export offsets of streams group in CSV format.
You must choose one of the following
offset specifications: --to-datetime,
--by-duration, --to-earliest, --to-
latest, --shift-by, --to-current, --to-offset.
To define the scope use --all-input-
topics or --input-topic. One scope
must be specified.
--input-topic <String: topic> The input topic whose committed offset
should be deleted, exported or reset. In
`reset-offsets` or `export-offsets` case, partitions can
be specified using this format:
`topic1:0,1,2`, where 0,1,2 are the
partition to be included in the
process. Multiple input topics can
be specified. Supported operations:
delete-offsets, reset-offsets, export-offsets.
--shift-by <Long: number-of-offsets> Reset or export offsets shifting current offset
by 'n', where 'n' can be positive or
negative.
--to-current Reset or export offsets to current offset.
--to-datetime <String: datetime> Reset or export offsets to offset from datetime.
Format: 'YYYY-MM-DDThh:mm:ss.sss'
--to-earliest Reset or export offsets to earliest offset.
--to-latest Reset or export offsets to latest offset.
--to-offset <Long: offset> Reset or export offsets to a specific offset.
Kafka protocol changes
A new RPC DescribeGroupsGeneric is introduced. Modern groups have type-specific RPCs like ConsumerGroupDescribe and ShareGroupDescribe , while classic groups have DescribeGroups . Each of the type-specific RPCs is specialised with type-specific information, and the new RPC is type-agnostic.
Access control
This table gives the ACLs required for the new RPCs.
| RPC | Operation | Resource |
|---|---|---|
DescribeGroupsGeneric | Read | Group |
DescribeGroupsGeneric RPC
The new DescribeGroupsGeneric RPC can be used to describe groups of all kinds. The response does not include type-specific information such as the list of members. It is intended primarily for use by tools which need to discover and validate group type efficiently.
Request schema
{
"apiKey": NNN,
"type": "request",
"listeners": ["broker"],
"name": "DescribeGroupsGenericRequest",
// Version 0 is the initial version (KIP-1359).
"validVersions": "0",
"flexibleVersions": "0+",
"headerVersions": {
"0+": "3"
},
"fields": [
{ "name": "GroupIds", "type": "[]string", "versions": "0+", "entityType": "groupId",
"about": "The ids of the groups to describe." },
{ "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+",
"about": "Whether to include authorized operations." }
]
}
Response schema
{
"apiKey": NNN,
"type": "response",
"name": "DescribeGroupsGenericResponse",
// Version 0 is the initial version (KIP-1359).
"validVersions": "0",
"flexibleVersions": "0+",
"headerVersions": {
"0+": "1"
},
// Supported errors:
// - GROUP_AUTHORIZATION_FAILED (version 0+)
// - NOT_COORDINATOR (version 0+)
// - COORDINATOR_NOT_AVAILABLE (version 0+)
// - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
// - INVALID_GROUP_ID (version 0+)
// - GROUP_ID_NOT_FOUND (version 0+)
// - INVALID_REQUEST (version 0+)
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
"about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "Groups", "type": "[]DescribedGroup", "versions": "0+",
"about": "Each described group.",
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The describe error, or 0 if there was no error." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The top-level error message, or null if there was no error." },
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The group ID string." },
{ "name": "GroupState", "type": "string", "versions": "0+",
"about": "The group state string, or the empty string." },
{ "name": "GroupType", "type": "string", "versions": "0+",
"about": "The group type name." }
{ "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
"about": "32-bit bitfield to represent authorized operations for this group." }
]
}
]
}
Compatibility, Deprecation, and Migration Plan
The users of --reset-offsets --export will be encouraged to move to the new --export-offsets option before the old option is removed in Apache Kafka 5.0.
Test Plan
Existing unit tests for resetting offsets using the existing tools will be converted to the new options. Tests will be added to the admin client changes.
Rejected Alternatives
We could change the optionality of the flags on the command-line tools instead. For example:
- We could permit --dry-run, --execute or --export as separate options. It would still be permitted to use --export with either of --dry-run or --execute, but using it by itself would also be allowed. The advantage of this is that there would be no need for users to migrate.
I also considered alternatives in the Kafka protocol.
- I could have enhanced the DescribeGroups RPC which is used for describing classic groups. However, that contains a lot of response fields which are specific to classic groups. As a result, I felt this was not suitable.
- I could have used ListGroups RPC instead. This has the downside that it must ask all brokers to list their groups and then combine their results, which is a very expensive operation.