THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
In order to fill in the gaps which have identified in the Group API, we propose to explicitly put the `ProtocolType` and the `ProtocolName` in all the requests and responses which contain the `Metadata` and/or the `Assignment`. Namely, we propose to add it them in the `JoinGroupResponse`, the `SyncGroupRequest` and the `SyncGroupResponse`. See Public Interfaces for details.
With this information available, we propose to extend the `GroupCoordinator` and the `AbstractCoordinator` to verify it them and to error out if it is they are not consistent with the Protocol Type and the Protocol Name of the group. This is not strictly required but contribute to increasing the robustness of the Group API. For instance, it could help to catch errors when a new client in implemented or when changes are made in the `GroupCoordinator`.
In the `GroupCoordinator`, during the handling of the `SyncGroupRequest`, we propose to return the `INCONSISTENT_GROUP_PROTOCOL` error if the consumer type provided by the client do not correspond to the consumer type and/or name of the group. For older version, the verification would be omitted.
In the `AbstractCoordinator`, during the handling of the `JoinGroupResponse` and the `SyncGroupResponse`, we propose to fail the future with the `INCONSISTENT_GROUP_PROTOCOL` error if the consumer type and/or name received is consistent with the consumer type known. It would behave similarly to failing to parse the metadata or the assignment.
...
| Code Block | ||||||
|---|---|---|---|---|---|---|
| ||||||
{
"apiKey": 14,
"type": "request",
"name": "SyncGroupRequest",
// Versions 1 and 2 are the same as version 0.
//
// Starting from version 3, we add a new field called groupInstanceId to indicate member identity across restarts.
//
// Version 4 is the first flexible version.
//
// Version 5 adds the Protocol Type.
"validVersions": "0-5",
"flexibleVersions": "4+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The unique group identifier." },
{ "name": "GenerationId", "type": "int32", "versions": "0+",
"about": "The generation of the group." },
{ "name": "MemberId", "type": "string", "versions": "0+",
"about": "The member ID assigned by the group." },
{ "name": "GroupInstanceId", "type": "string", "versions": "3+",
"nullableVersions": "3+", "default": "null",
"about": "The unique identifier of the consumer instance provided by end user." },
// New Field
{ "name": "ProtocolType", "type": "string", "versions": "5+",
"nullableVersions": "5+", "default": "null", "ignorable": true,
"about": "The unique name the for class of protocols implemented by the group we want to join." },
// New Field
{ "name": "AssignmentsProtocolName", "type": "[]SyncGroupRequestAssignmentstring", "versions": "05+",
"aboutnullableVersions": "5+", "Each assignment.default": "null", "fieldsignorable": [true,
{ "nameabout": "MemberId", "The group protocol name." },
{ "name": "Assignments", "type": "string[]SyncGroupRequestAssignment", "versions": "0+",
"about": "The ID of the member to assign." },Each assignment.", "fields": [
{ "name": "AssignmentMemberId", "type": "bytesstring", "versions": "0+",
"about": "The ID of the member assignmentto assign." },
] { "name": "Assignment", "type": "bytes", "versions": "0+",
"about": "The member assignment." }
]}
]
} |
SyncGroupResponse
| Code Block | ||||||
|---|---|---|---|---|---|---|
| ||||||
{
"apiKey": 14,
"type": "response",
"name": "SyncGroupResponse",
// Version 1 adds throttle time.
//
// Starting in version 2, on quota violation, brokers send out responses before throttling.
//
// Starting from version 3, syncGroupRequest supports a new field called groupInstanceId to indicate member identity across restarts.
//
// Version 4 is the first flexible version.
//
// Version 5 adds the Protocol Type.
"validVersions": "0-5",
"flexibleVersions": "4+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "1+", "ignorable": true,
"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": ": "ErrorCodeErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
// New Fields
{ "name": "ProtocolType", "type": "string", "versions": "5+",
"nullableVersions": "5+", "typedefault": "int16null", "versionsignorable": "0+"true,
"about": "The error code, or 0 if there was no errorunique name the for class of protocols implemented by the group we want to join." },
// New Field
{ "name": "ProtocolTypeProtocolName", "type": "string", "versions": "5+",
"nullableVersions": "5+", "default": "null", "ignorable": true,
"about": "The unique name the for class of protocols implemented by the group we want to joinprotocol name." },
{ "name": "Assignment", "type": "bytes", "versions": "0+",
"about": "The member assignment." }
]
} |
...