DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
All non-assignment metadata as defined in the
ConsumerGroupHeartbeatRequestsuch as the group ID, member ID, member epoch, instance ID, rack ID and rebalance timeout.The currently assigned active, standby and warm-up tasks. Each task is identified by a subtopology ID (which can be an arbitrary string in protocol for future extendability, even though subtopologies are currently numbered) and a partition number.
The process ID to identify multiple Streams clients running in the same process.
The user-defined endpoint to be used for running interactive queries.
A unique ID of the processing topology. The ID is defined by the client, and will be hash of the topology representation by default.
For each task, the current sum of task changelog changelog offsets and task changelog end offsets. These are used to determine whether a warm-up task is below
acceptable.recovery.lagand can be used to optimize the assignment in the assignor. For more details on these offsets see "Cumulative task changelog offsets and task changelog end offsets" section belowA flag to request shutdown of the whole application (only if the member epoch is the group leave epoch).
...
The group epoch will be bumped every time the topology is updated through the
StreamsGroupInitializeAPI.The partition metadata for the group, which is tracked by the group coordinator, is the partition metadata of all input topics (i.e., user source topics and internal repartition topics we read from).
The group epoch is bumped every time a member:
When a client with an assigned warm-up task reports a task changelog offset and task changelog end offset whose difference is less that
acceptable.recovery.lag.Updates its topology ID, rack ID, client tags or process ID. Note: Typically, these do not change within the lifetime of a Streams client, so this only happens when a member with static membership rejoins with an updated configuration.
...
For each processing (active) task, the sum of the offsets last fully replicated to the broker in the changelog topic of each local state. Under at-least-once guarantees, this is going to correspond to the high watermark, under exactly-once semantics, it’s going to be the last stable offset.
For each restoring (active, standby or warm-up), task the sum of the positions in the changelog topic and checkpointed to the local state directory.
For each dormant task (task which is not owned, but which has state locally on disk), the sum of the checkpointed positions in the changelog topic present in its state directory. The client only reports the offsets for dormant tasks that we manage to acquire the lock to the state directory for.
Cumulative task changelog end offsets
Similarly, the client reports the sum of the end offsets in all changelog topics for the currently owned tasks, if available. This simplifies the broker-side task assignment, since the broker doesn’t need to fetch the current end-offsets. Specifically:
For active processing tasks, the cumulative task changelog end offset is the same as the cumulative task changelog offset
For all restoring tasks (active, standby or warm-up), the cumulative end-offset is the sum of the last end offsets cached by the restore consumer. If an end offset of a topic partition is unknown, no end-offset is reported.
For dormant standby tasks, no end-offset is reported.
Using cumulative task changelog offsets and cumulative task changelog end offsets
By reporting cumulative task changelog offsets and cumulative task changelog end offsets instead of the cumulative task lag, we can determine most task lags in the group coordinator, without the group coordinator or the streams client having to do additional requests. Even if the end offsets of the topic partition is not known on that client, as long as it is known on another client in the group. However, it may happen that no end offset is known in the group coordinator - for example, for tasks that are dormant on at least one client, and no client currently owns that task as active, standby or warm-up tasks. In these cases, the task with unknown end-offsets are never considered to be caught-up, however, the cumulative task changelog offset can still be used when deciding on where to place a new active, standby or warm-up task - typically, by selecting the client with the maximal cumulative task changelog offset that fulfils all other requirements.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker"],
"name": "StreamsGroupHeartbeatRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The group identifier." },
{ "name": "MemberId", "type": "string", "versions": "0+",
"about": "The member id generated by the coordinator. The member id must be kept during the entire lifetime of the member." },
{ "name": "MemberEpoch", "type": "int32", "versions": "0+",
"about": "The current member epoch; 0 to join the group; -1 to leave the group; -2 to indicate that the static member will rejoin." },
{ "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "null if not provided or if it didn't change since the last heartbeat; the instance Id otherwise." },
{ "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "null if not provided or if it didn't change since the last heartbeat; the rack ID of consumer otherwise." },
{ "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
"about": "-1 if it didn't change since the last heartbeat; the maximum time in milliseconds that the coordinator will wait on the member to revoke its partitions otherwise." },
{ "name": "TopologyId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": null,
"about": "The ID of the topology. Only sent when memberEpoch = 0, must be non-empty. Null otherwise. Used to check if topology corresponds to server-topology. " },
{ "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Currently owned active tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Currently owned standby tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Currently owned warm-up tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "ProcessId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Identity of the streams instance that may have multiple consumers. Null if unchanged since last heartbeat." },
{ "name": "UserEndpoint", "type": "Endpoint", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "User-defined endpoint for Interactive Queries. Null if unchanged since last heartbeat." },
{ "name": "ClientTags", "type": "[]KeyValue", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Used for rack-aware assignment algorithm. Null if unchanged since last heartbeat." },
{ "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Cumulative changelog offsets for tasks. Only updated when a warm -up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
{ "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Cumulative changelog end-offsets for tasks. Only updated when a warm -up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
{ "name": "ShutdownApplication", "type": "bool", "versions": "0+", "default": false,
"about": "Whether all Streams clients in the group should shut down. Can only be defined if memberEpoch = -1" }
],
"commonStructs": [
{ "name": "Endpoint", "versions": "0+", "fields": [
{ "name": "Host", "type": "string", "versions": "0+",
"about": "host of the endpoint" },
{ "name": "Port", "type": "int32", "versions": "0+",
"about": "port of the endpoint" }
]},
{ "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": "TaskOffset", "versions": "0+", "fields": [
{ "name": "Subtopology", "type": "string", "versions": "0+",
"about": "The sub-topology identifier." },
{ "name": "Partition", "type": "int32", "versions": "0+",
"about": "The partition." },
{ "name": "Offset", "type": "int64", "versions": "0+",
"about": "The offset." }
]},
{ "name": "TaskIds", "versions": "0+", "fields": [
{ "name": "Subtopology", "type": "string", "versions": "0+",
"about": "The sub-topology identifier." },
{ "name": "Partitions", "type": "[]int32", "versions": "0+",
"about": "The partitions of the input topics processed by this member." }
]}
]
} |
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
{
"apiKey": TBD,
"type": "response",
"name": "StreamsGroupHeartbeatResponse",
"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+)
// - UNKNOWN_MEMBER_ID (version 0+)
// - FENCED_MEMBER_EPOCH (version 0+)
// - UNRELEASED_INSTANCE_ID (version 0+)
// - GROUP_MAX_SIZE_REACHED (version 0+)
"fields": [
// Same as consumer group heart beat
{ "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": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The top-level error code, 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": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The member id generated by the coordinator. Only provided when the member joins with MemberEpoch == 0." },
{ "name": "MemberEpoch", "type": "int32", "versions": "0+",
"about": "The member epoch." },
{ "name": "HeartbeatIntervalMs", "type": "int32", "versions": "0+",
"about": "The heartbeat interval in milliseconds." },
{ "name": "AcceptableRecoveryLag", "type": "int32", "versions": "0+",
"about": "The maximal lag a warm-up task can have to be considered caught-up." },
{ "name": "TaskOffsetIntervalMs", "type": "int32", "versions": "0+",
"about": "The interval in which the task changelog task offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },
// Topology updating
{ "name": "ShouldInitializeTopology", "type": "bool", "versions": "0+", "default": false,
"about": "true if this streams application should initialize the topology on the broker" },
{ "name": "GroupTopologyId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": null,
"about": "The current ID of the topology for the group. Null if unchanged since last heartbeat, empty if group is still initializing." },
{ "name": "Status", "type": "[]Status", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Indicate zero or more status for the group. Null if unchanged since last heartbeat." },
// The streams app knows which partitions to fetch from given this information
{ "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Assigned active tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Assigned standby tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Assigned warm-up tasks for this client. Null if unchanged since last heartbeat." },
// IQ-related information
{ "name": "PartitionsByUserEndpoint", "type": "[]EndpointToPartitions", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Global assignment information used for IQ. Null if unchanged since last heartbeat." ,
"fields": [
{ "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
"about": "User-defined endpoint to connect to the node" },
{ "name": "Partitions", "type": "[]TopicPartition", "versions": "0+",
"about": "All partitions available on the node" }
]
}
],
"commonStructs": [
{ "name": "Status", "versions": "0+", "fields": [
// Possible status codes
// 0 - GROUP_INITIALIZING - The group is not initialized.
// 1 - INCONSISTENT_TOPOLOGY - The topology ID supplied is inconsistent with the topology for this streams group.
// 2 - MISSING_SOURCE_TOPICS - One or more source topics do not exist. Missing topics are indicated in the StatusDetail.
// 3 - MISSING_INTERNAL_TOPICS - One or more internal topics do not exist. Missing topics are indicated in the StatusDetail.
// 4 - SHUTDOWN_APPLICATION - A client requested the shutdown of the whole application.
{ "name": "StatusCode", "type": "int8", "versions": "0+",
"about": "A code to indicate that a particular status is active for the group membership" },
{ "name": "StatusDetail", "type": "string", "versions": "0+",
"about": "A string representation of the status." }
]},
{ "name": "TopicPartition", "versions": "0+", "fields": [
{ "name": "Topic", "type": "string", "versions": "0+",
"about": "topic name" },
{ "name": "Partitions", "type": "[]int32", "versions": "0+",
"about": "partitions" }
]},
{ "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": "Endpoint", "versions": "0+", "fields": [
{ "name": "Host", "type": "string", "versions": "0+",
"about": "host of the endpoint" },
{ "name": "Port", "type": "int32", "versions": "0+",
"about": "port of the endpoint" }
]}
]
} |
...
| 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": "TopologyId", "type": "string", "versions": "0+",
"about": "The ID of the currently initialized topology for this group." },
{ "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": "RepartitionSinkTopics", "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 on the client." },
{ "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 changelog offsets for tasks." },
{ "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+",
"about": "Cumulative changelog 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."
}
]}
]
} |
...
- The member metadata record, keyed by the group ID and the member ID, stores all relevant metadata that is sent by a member to the coordinator in a heartbeat request. A notable exception are task changelog offsets and task lag changelog end offsets, which are not persisted, as they are constantly changing.
- The group metadata, keyed by the group ID, contains the group epoch.
...
Name | Type | Default | Doc |
|---|---|---|---|
group.streams.session.timeout.ms | int | 45s | The timeout to detect client failures when using the streams group protocol. |
group.streams.min.session.timeout.ms | int | 45s | The minimum session timeout. |
group.streams.max.session.timeout.ms | int | 60s | The maximum session timeout. |
group.streams.heartbeat.interval.ms | int | 5s | The heartbeat interval given to the members. |
group.streams.min.heartbeat.interval.ms | int | 5s | The minimum heartbeat interval. |
group.streams.max.heartbeat.interval.ms | int | 15s | The maximum heartbeat interval. |
group.streams.max.size | int | MaxValue | The maximum number of streams clients that a single streams group can accommodate. |
group.streams.acceptable.recovery.lag | long | 10’000 | The maximum acceptable lag (number of offsets to catch up) for a client to be considered caught-up enough to receive an active task assignment. |
group.streams.num.warmup.replicas | int | 2 | The maximum number of warmup replicas. |
group.streams.max.warmup.replicas | int | 4 | Maximum for dynamic configurations of the warmup replica configuration |
group.streams.num.standby.replicas | int | 0 | The number of standby replicas for each task. |
group.streams.max.standby.replicas | int | 2 | Maximum for dynamic configurations of the standby replica configuration |
group.streams.task.offset.interval.ms | int | 60s | The interval in which the task changelog task offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed. |
group.streams.min.task.offset.interval.ms | int | 15s | Minimum for dynamic configurations of the task offset interval. Used to restrict users of the cluster from sending the task changelog offsets too often. |
group.streams.assignor.name | string | null | The name of the task assignor used for all streams groups. Can be |
...
Name | Type | Default | Doc |
|---|---|---|---|
group.streams.session.timeout.ms | int | 45s | The timeout to detect client failures when using the streams group protocol. |
group.streams.heartbeat.interval.ms | int | 5s | The heartbeat interval given to the members. |
group.streams.acceptable.recovery.lag | long | 10’000 | The maximum acceptable lag (number of offsets to catch up) for a client to be considered caught-up enough to receive an active task assignment. |
group.streams.maxnum.warmup.replicas | int | 2 | The maximum number of warmup replicas. |
group.streams.num.standby.replicas | int | 0 | The number of standby replicas for each task. |
group.streams.task.offset.interval.ms | int | 60s | The interval in which the task changelog task offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed. |
group.streams.assignor.name | string | null | The name of the task assignor used for this streams groups. Can be |
...
| Code Block | ||||||
|---|---|---|---|---|---|---|
| ||||||
/**
* A detailed description of a single member in the group.
*/
public class StreamsGroupMemberDescription {
public StreamsGroupMemberDescription(
final String memberId,
final Optional<String> instanceId,
final String clientId,
final String clientHost,
final String topologyId,
final String processId,
final Map<String, String> clientTags,
final List<TaskOffset> taskOffsets,
final List<TaskOffset> taskEndOffsets,
final StreamsGroupMemberAssignment assignment,
final Optional<StreamsGroupMemberAssignment> targetAssignment);
/**
* The id of the group member.
*/
public String memberId();
/**
* The id of the instance, if available.
*/
public Optional<String> instanceId();
/**
* The client id of the group member.
*/
public String clientId();
/**
* The host of the group member.
*/
public String clientHost();
/**
* The id of the topology present on the client.
*/
public String topologyId();
/**
* Identity of the streams instance that may have multiple clients.
*/
public String processId();
/**
* Used for rack-aware assignment algorithm.
*/
public Map<String, String> clientTags();
/**
* Cumulative offsets for tasks.
*/
public List<TaskOffset> taskOffsets();
/**
* Cumulative task changelog end offsets for tasks.
*/
public List<TaskOffset> taskEndOffsets();
/**
* The current assignment.
*/
public StreamsGroupMemberAssignment assignment();
/**
* The target assignment.
*/
public Optional<StreamsGroupMemberAssignment> targetAssignment();
/**
* The cumulative offset for one task.
*/
public static class TaskOffset {
public TaskOffset(final String subtopology, final int partition, final long offset);
/**
* The subtopology identifier.
*/
public String subtopology();
/**
* The partition of the task.
*/
public int partition();
/**
* The cumulative offset (sum of offsets in all input partitions).
*/
public long offset();
}
} |
...