THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
Two new control record types are introduced: SHARE_CHECKPOINT (5) and SHARE_DELTA (6). They are written into separate message sets with the Control flag set. This flag indicates that the records are not intended for application consumption. Indeed, these message sets are not returned to any consumers at all since they are just intended for the share-partition leader.
When a control record is written as a result of an operation such as a ShareAcknowledge RPC, the control record must be written and fully replicated before the RPC response is sent.
SHARE_CHECKPOINT
A SHARE_CHECKPOINT record contains a complete checkpoint of the share-partition state. It contains:
...
| Method signature | Description |
|---|---|
void onComplete(Map<TopicIdPartition, Optional<KafkaException>> results, Set<OffsetAndMetadata>> offsets, Exception exception) | A callback method the user can implement to provide asynchronous handling of request completion. Parameters: offsets - A map of the offsets that this callback applies to. exception - The exception thrown during processing of the request, or null if the acknowledgement completed successfully. Exceptions: WakeupException - if KafkaShareConsumer.wakeup() is called. InterruptException - if the calling thread is interrupted. AuthorizationException - if not authorized to the topic or group. KafkaException - for any other unrecoverable errors. |
AcknowledgeType
The new org.apache.kafka.clients.consumer.AcknowledgeType enum distinguishes between the types of acknowledgement for a record consumer using a share group.
...
| Code Block | ||
|---|---|---|
| ||
package org.apache.kafka.client.admin;
import org.apache.kafka.common.GroupType;
/**
* Options for {@link Admin#listGroups(ListGroupsOptions)}.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
public class ListGroupsOptions extends AbstractOptions<ListGroupsOptions> {
/**
* If types is set, only groups of these types will be returned. Otherwise, all groups are returned.
*/
public ListGroupsOptions types(Set<GroupType> statestypes);
/**
* Return the list of types that are requested or empty if no types have been specified.
*/
public Set<GroupType> types();
} |
...
| Code Block |
|---|
{
"apiKey": NN,
"type": "request",
"listeners": ["broker"],
"name": "ShareFetchRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null", "entityType": "groupId",
"about": "null if not provided or if it didn't change since the last fetch; the group identifier otherwise." },
{ "name": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+",
"about": "The member ID." },
{ "name": "ShareSessionEpoch", "type": "int32", "versions": "0+",
"about": "The current share session epoch: 0 to open a share session; -1 to close it; otherwise increments for consecutive requests." },
{ "name": "MaxWaitMs", "type": "int32", "versions": "0+",
"about": "The maximum time in milliseconds to wait for the response." },
{ "name": "MinBytes", "type": "int32", "versions": "0+",
"about": "The minimum bytes to accumulate in the response." },
{ "name": "MaxBytes", "type": "int32", "versions": "0+", "default": "0x7fffffff", "ignorable": true,
"about": "The maximum bytes to fetch. See KIP-74 for cases where this limit may not be honored." },
{ "name": "Topics", "type": "[]FetchTopic", "versions": "0+",
"about": "The topics to fetch.", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "ignorable": true, "about": "The unique topic ID."},
{ "name": "Partitions", "type": "[]FetchPartition", "versions": "0+",
"about": "The partitions to fetch.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "CurrentLeaderEpoch", "type": "int32", "versions": "0+", "default": "-1", "ignorable": true,
"about": "The current leader epoch of the partition." },
{ "name": "PartitionMaxBytes", "type": "int32", "versions": "0+",
"about": "The maximum bytes to fetch from this partition. See KIP-74 for cases where this limit may not be honored." },
{ "name": "AcknowledgementBatches", "type": "[]AcknowledgementBatch", "versions": "0+",
"about": "Record batches to acknowledge.", "fields": [
{ "name": "StartOffset", "type": "int64", "versions": "0+",
"about": "Start offset of batch of records to acknowledge."},
{ "name": "LastOffset", "type": "int64", "versions": "0+",
"about": "Last offset (inclusive) of batch of records to acknowledge."},
{ "name": "GapOffsets", "type": "[]int64", "versions": "0+",
"about": "Array of offsets in this range which do not correspond to records."},
{ "name": "AcknowledgeType", "type": "int8", "versions": "0+", "default": "0",
"about": "The type of acknowledgement - 0:Accept,1:Release,2:Reject."}
]}
]},
{ "name": "ForgottenTopicsData", "type": "[]ForgottenTopic", "versions": "0+", "ignorable": false,
"about": "The partitions to remove from this share session.", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "ignorable": true, "about": "The unique topic ID."},
{ "name": "Partitions", "type": "[]int32", "versions": "0+",
"about": "The partitions indexes to forget." }
]}
]
} |
...
| Code Block |
|---|
{
"apiKey": NN,
"type": "request",
"listeners": ["broker"],
"name": "ShareAcknowledgeRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+",
"about": "The member ID." },
{ "name": "ShareSessionEpoch", "type": "int32", "versions": "0+",
"about": "The current share session epoch: 0 to open a share session; -1 to close it; otherwise increments for consecutive requests." },
{ "name": "Topics", "type": "[]AcknowledgeTopic", "versions": "0+",
"about": "The topics containing records to acknowledge.", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "Partitions", "type": "[]AcknowledgePartition", "versions": "0+",
"about": "The partitions containing records to acknowledge.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "AcknowledgementBatches", "type": "[]AcknowledgementBatch", "versions": "0+",
"about": "Record batches to acknowledge.", "fields": [
{ "name": "StartOffset", "type": "int64", "versions": "0+",
"about": "Start offset of batch of records to acknowledge."},
{ "name": "LastOffset", "type": "int64", "versions": "0+",
"about": "Last offset (inclusive) of batch of records to acknowledge."},
{ "name": "GapOffsets", "type": "[]int64", "versions": "0+",
"about": "Array of offsets in this range which do not correspond to records."},
{ "name": "AcknowledgeType", "type": "int8", "versions": "0+", "default": "0",
"about": "The type of acknowledgement - 0:Accept,1:Release,2:Reject."}
]}
]}
]}
]
} |
...
| Code Block |
|---|
{
"type": "data",
"name": "ShareDeltaValue",
"validVersions": "0",
"flexibleVersions": "none",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0",
"about": "The group identifier." },
{ "name": "CheckpointEpoch", "type": "uint16", "versions": "0",
"about": "The checkpoint epoch, increments with each checkpoint." },
{ "name": "States", "type": "[]State", "versions": "0", "fields": [
{ "name": "BaseOffset", "type": "int64", "versions": "0",
"about": "The base offset of this state batch." },
{ "name": "LastOffset", "type": "int64", "versions": "0",
"about": "The last offset of this state batch." },
{ "name": "State", "type": "int8", "versions": "0",
"about": "The state - 0:Available,2:Acked,4:Archived." },
{ "name": "DeliveryCount", "type": "int16", "versions": "0",
"about": "The delivery count." }
]}
]
} |
Index structure for locating share-partition state
More information needs to be added to describe how the index for locating the share-partition state is arranged.
Metrics
Broker Metrics
The following new broker metrics should be added:
...