DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
{
"apiKey": TBD,
"type": "response",
"name": "StreamsGroupDescribeResponse",
"validVersions": "0",
"flexibleVersions": "0+",
// Supported errors:
// - GROUP_AUTHORIZATION_FAILED (version 0+)
// - NOT_COORDINATOR (version 0+)
// - COORDINATOR_NOT_AVAILABLE (version 0+)
// - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
// - INVALID_REQUEST (version 0+)
// - INVALID_GROUP_ID (version 0+)
// - GROUP_ID_NOT_FOUND (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": "GroupEpoch", "type": "int32", "versions": "0+",
"about": "The group epoch." },
{ "name": "AssignmentEpoch", "type": "int32", "versions": "0+",
"about": "The assignment epoch." },
{ "name": "Topology", "type": "[]Subtopology", "versions": "0+",
"about": "The sub-topologies of the streams application.",
"fields": [
{ "name": "Subtopology", "type": "string", "versions": "0+",
"about": "String to uniquely identify the subtopology." },
{ "name": "SourceTopics", "type": "[]string", "versions": "0+",
"about": "The topics the topology reads from." },
{ "name": "SourceTopicRegex", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The regular expressions identifying topics the topology reads from. null if not provided." },
{ "name": "SinkTopicsRepartitionSinkTopics", "type": "[]string", "versions": "0+",
"about": "The topics the topology writes to." },
{ "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
"about": "The set of state changelog topics associated with this subtopology. Created automatically." },
{ "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
"about": "The set of source topics that are internally created repartition topics. Created automatically." }
]
},
{ "name": "Members", "type": "[]Member", "versions": "0+",
"about": "The members.",
"fields": [
{ "name": "MemberId", "type": "string", "versions": "0+",
"about": "The member ID." },
{ "name": "MemberEpoch", "type": "int32", "versions": "0+",
"about": "The member epoch." },
{ "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The member instance ID." },
{ "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The rack ID." },
{ "name": "ClientId", "type": "string", "versions": "0+",
"about": "The client ID." },
{ "name": "ClientHost", "type": "string", "versions": "0+",
"about": "The client host." },
{ "name": "TopologyId", "type": "string", "versions": "0+",
"about": "The ID of the topology. Must be non-empty." },
{ "name": "ProcessId", "type": "string", "versions": "0+",
"about": "Identity of the streams instance that may have multiple clients. " },
{ "name": "ClientTags", "type": "[]KeyValue", "versions": "0+",
"about": "Used for rack-aware assignment algorithm." },
{ "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+",
"about": "Cumulative offsets for tasks." },
{ "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+",
"about": "Cumulative end offsets for tasks." },
{ "name": "Assignment", "type": "Assignment", "versions": "0+",
"about": "The current assignment." },
{ "name": "TargetAssignment", "type": "Assignment", "versions": "0+",
"about": "The target assignment." }
]},
{ "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
"about": "32-bit bitfield to represent authorized operations for this group." }
]
}
],
"commonStructs": [
{ "name": "TaskOffset", "versions": "0+", "fields": [
{ "name": "Subtopology", "type": "string", "versions": "0+",
"about": "The subtopology identifier." },
{ "name": "Partition", "type": "int32", "versions": "0+",
"about": "The partition." },
{ "name": "Offset", "type": "int64", "versions": "0+",
"about": "The offset." }
]},
{ "name": "TopicPartitions", "versions": "0+", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+",
"about": "The topic ID." },
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]int32", "versions": "0+",
"about": "The partitions." }
]},
{ "name": "Assignment", "versions": "0+", "fields": [
{ "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+",
"about": "Active tasks for this client." },
{ "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+",
"about": "Standby tasks for this client." },
{ "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+",
"about": "Warm-up tasks for this client. " }
]},
{ "name": "TaskIds", "versions": "0+", "fields": [
{ "name": "Subtopology", "type": "string", "versions": "0+",
"about": "The subtopology identifier." },
{ "name": "Partitions", "type": "[]int32", "versions": "0+",
"about": "The partitions of the input topics processed by this member." }
]},
{ "name": "KeyValue", "versions": "0+", "fields": [
{ "name": "Key", "type": "string", "versions": "0+",
"about": "key of the config" },
{ "name": "Value", "type": "string", "versions": "0+",
"about": "value of the config" }
]},
{ "name": "TopicInfo", "versions": "0+", "fields": [
{ "name": "Name", "type": "string", "versions": "0+",
"about": "The name of the topic." },
{ "name": "Partitions", "type": "int32", "versions": "0+",
"about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
{ "name": "TopicConfigs", "type": "[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Topic-level configurations as key-value pairs."
}
]}
]
} |
...
Note that both the rack-aware task assignor and customizable client-side task assignment may be introduced into the new protocol in follow-up KIPs (see Future Work section). In this case, the latter 5 configuration options would not be deprecated in the future major version.
Admin API
Command-line tools
kafka-streams-groups.sh
A new tool called kafka-streams-groups.sh is added for working with streams groups. It has the following options:
Option | Description |
|---|---|
--version | Display Kafka version. |
--all-input-topics | Reset all input topic in the reset offset process, as stored in the topology metadata on the broker. |
--input-topics <String: topics> | The input topic that should be included in the reset offset process. |
--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 share 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. |
--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. |
--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. |
Here are some examples.
To display a list of all streams groups:
$ kafka-streams-groups.sh --bootstrap-server localhost:9092 --list
To delete the information for topic T1 from inactive streams group S1 , which essentially resets the consumption of this topic in the share group:
$ kafka-streams-groups.sh --bootstrap-server localhost:9092 --group S1 --delete-offsets
To set the starting offset for consuming topic T1 in 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 1999-12-31T23:57:00.000 --execute
kafka-streams-application-reset
...