[
https://issues.apache.org/jira/browse/KAFKA-20624?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
TengYao Chi updated KAFKA-20624:
--------------------------------
Description:
streamsGroupDescribe gains includeTopologyDescription flag.
attachTopologyDescriptions: call plugin.getTopology only when
StoredTopologyEpoch == currentTopologyEpoch, otherwise report NOT_STORED.
Status constants: NOT_REQUESTED (0), NOT_STORED (1), ERROR (2), AVAILABLE (3).
GroupCoordinator interface update. Broker wiring: KafkaApis.scala (new handler
+ describe flag plumbing), BrokerServer.scala (plugin instantiation +
configuration), RequestConvertToJson. GroupCoordinatorServiceTest
(describe-path tests), KafkaApisTest, RequestQuotaTest,
AuthorizerIntegrationTest, checkstyle/suppressions.xml.
Wire the streams-group topology description plugin into broker-side request
handling, completing the KAFKA-20618 broker work that KAFKA-20620 (schema +
stubs) and KAFKA-20623 (coordinator-side write/heartbeat/delete) left for
this ticket.
Two paths are wired:
1. Describe path (read). streamsGroupDescribe is extended with an
IncludeTopologyDescription request flag. When the flag is set and the
persisted StoredDescriptionTopologyEpoch matches the group's current
topology epoch, a new StreamsGroupTopologyDescriptionManager.
attachTopologyDescriptions building block calls
plugin.getTopology(groupId, topologyEpoch) per group and populates the
response's topology field. Each DescribedGroup carries a
topologyDescriptionStatus byte: NOT_REQUESTED (0) when the client did
not ask for it, NOT_STORED (1) when the plugin has no row at the
requested epoch (or the epoch mismatched), ERROR (2) when the plugin
call failed, AVAILABLE (3) when the topology is attached. Chain
assembly lives on GroupCoordinatorService.streamsGroupDescribe; the
manager exposes per-group plugin invocation as a building block
(mirroring invokeSetTopology / invokeDeleteTopologies).
2. Update path (write — KafkaApis wiring).
KafkaApis.handleStreamsGroupTopologyDescriptionUpdate currently lives as
a placeholder stub that always returns UNSUPPORTED_VERSION (scaffolding
from KAFKA-20620). Replace it with the real handler: gate on
isStreamsGroupProtocolEnabled (returns UNSUPPORTED_VERSION when the
streams group protocol is disabled), authorize READ on GROUP for the
request's groupId (returns GROUP_AUTHORIZATION_FAILED if denied — per
KIP-1331, "like offset commits, we don't consider this a modification
of the GROUP, which allows deploying apps with READ ACLs"), then call
groupCoordinator.streamsGroupTopologyDescriptionUpdate(...) and send
the result via requestHelper.sendMaybeThrottle. Same shape as the
sibling handleStreamsGroupHeartbeat. The coordinator-side method was
added in KAFKA-20623 split 2 (#22552); only the KafkaApis side is
missing — flagged on the split 2 review.
Broker wiring (shared). BrokerServer.scala instantiates the plugin via
config.getConfiguredInstance(...) and threads it into the
GroupCoordinatorService builder (verify the describe path doesn't need
additional wiring beyond what KAFKA-20623 already added).
RequestConvertToJson adds entries for StreamsGroupTopologyDescriptionUpdate
Request/Response and for the new topology-description fields on the
describe response so request/response logging tools can render them.
was:
streamsGroupDescribe gains includeTopologyDescription flag.
attachTopologyDescriptions: call plugin.getTopology only when
StoredTopologyEpoch == currentTopologyEpoch, otherwise report NOT_STORED.
Status constants: NOT_REQUESTED (0), NOT_STORED (1), ERROR (2), AVAILABLE (3).
GroupCoordinator interface update. Broker wiring: KafkaApis.scala (new handler
+ describe flag plumbing), BrokerServer.scala (plugin instantiation +
configuration), RequestConvertToJson. GroupCoordinatorServiceTest
(describe-path tests), KafkaApisTest, RequestQuotaTest,
AuthorizerIntegrationTest, checkstyle/suppressions.xml.
Wire the streams-group topology description plugin into broker-side request
handling, completing the KAFKA-20618 broker work that KAFKA-20620 (schema +
stubs) and KAFKA-20623 (coordinator-side write/heartbeat/delete) left for
this ticket.
Two paths are wired:
1. Describe path (read). streamsGroupDescribe is extended with an
IncludeTopologyDescription request flag. When the flag is set and the
persisted StoredDescriptionTopologyEpoch matches the group's current
topology epoch, a new StreamsGroupTopologyDescriptionManager.
attachTopologyDescriptions building block calls
plugin.getTopology(groupId, topologyEpoch) per group and populates the
response's topology field. Each DescribedGroup carries a
topologyDescriptionStatus byte: NOT_REQUESTED (0) when the client did
not ask for it, NOT_STORED (1) when the plugin has no row at the
requested epoch (or the epoch mismatched), ERROR (2) when the plugin
call failed, AVAILABLE (3) when the topology is attached. Chain
assembly lives on GroupCoordinatorService.streamsGroupDescribe; the
manager exposes per-group plugin invocation as a building block
(mirroring invokeSetTopology / invokeDeleteTopologies).
2. Update path (write — KafkaApis wiring).
KafkaApis.handleStreamsGroupTopologyDescriptionUpdate currently lives
as a placeholder stub that always returns UNSUPPORTED_VERSION
(scaffolding from KAFKA-20620). Replace it with the real handler: gate
on isStreamsGroupProtocolEnabled, authorize WRITE on GROUP for the
request's groupId (returning GROUP_AUTHORIZATION_FAILED if denied, per
KIP-1331), then call groupCoordinator.
streamsGroupTopologyDescriptionUpdate(...) and send the result via
requestHelper.sendMaybeThrottle. The coordinator-side method was added
in KAFKA-20623 split 2 (#22552); only the KafkaApis side is missing —
flagged on the split 2 review.
Broker wiring (shared). BrokerServer.scala instantiates the plugin via
config.getConfiguredInstance(...) and threads it into the
GroupCoordinatorService builder (verify the describe path doesn't need
additional wiring beyond what KAFKA-20623 already added).
RequestConvertToJson adds entries for StreamsGroupTopologyDescriptionUpdate
Request/Response and for the new topology-description fields on the
describe response so request/response logging tools can render them.
> Wire topology description plugin into GroupCoordinatorService — read path
> (describe) and broker wiring
> ------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20624
> URL: https://issues.apache.org/jira/browse/KAFKA-20624
> Project: Kafka
> Issue Type: Sub-task
> Components: streams
> Reporter: Alieh Saeedi
> Assignee: TengYao Chi
> Priority: Major
> Fix For: 4.4.0
>
>
> streamsGroupDescribe gains includeTopologyDescription flag.
> attachTopologyDescriptions: call plugin.getTopology only when
> StoredTopologyEpoch == currentTopologyEpoch, otherwise report NOT_STORED.
> Status constants: NOT_REQUESTED (0), NOT_STORED (1), ERROR (2), AVAILABLE
> (3). GroupCoordinator interface update. Broker wiring: KafkaApis.scala (new
> handler + describe flag plumbing), BrokerServer.scala (plugin instantiation +
> configuration), RequestConvertToJson. GroupCoordinatorServiceTest
> (describe-path tests), KafkaApisTest, RequestQuotaTest,
> AuthorizerIntegrationTest, checkstyle/suppressions.xml.
> Wire the streams-group topology description plugin into broker-side request
> handling, completing the KAFKA-20618 broker work that KAFKA-20620 (schema +
> stubs) and KAFKA-20623 (coordinator-side write/heartbeat/delete) left for
> this ticket.
> Two paths are wired:
> 1. Describe path (read). streamsGroupDescribe is extended with an
> IncludeTopologyDescription request flag. When the flag is set and the
> persisted StoredDescriptionTopologyEpoch matches the group's current
> topology epoch, a new StreamsGroupTopologyDescriptionManager.
> attachTopologyDescriptions building block calls
> plugin.getTopology(groupId, topologyEpoch) per group and populates the
> response's topology field. Each DescribedGroup carries a
> topologyDescriptionStatus byte: NOT_REQUESTED (0) when the client did
> not ask for it, NOT_STORED (1) when the plugin has no row at the
> requested epoch (or the epoch mismatched), ERROR (2) when the plugin
> call failed, AVAILABLE (3) when the topology is attached. Chain
> assembly lives on GroupCoordinatorService.streamsGroupDescribe; the
> manager exposes per-group plugin invocation as a building block
> (mirroring invokeSetTopology / invokeDeleteTopologies).
> 2. Update path (write — KafkaApis wiring).
> KafkaApis.handleStreamsGroupTopologyDescriptionUpdate currently lives as
> a placeholder stub that always returns UNSUPPORTED_VERSION (scaffolding
> from KAFKA-20620). Replace it with the real handler: gate on
> isStreamsGroupProtocolEnabled (returns UNSUPPORTED_VERSION when the
> streams group protocol is disabled), authorize READ on GROUP for the
> request's groupId (returns GROUP_AUTHORIZATION_FAILED if denied — per
> KIP-1331, "like offset commits, we don't consider this a modification
> of the GROUP, which allows deploying apps with READ ACLs"), then call
> groupCoordinator.streamsGroupTopologyDescriptionUpdate(...) and send
> the result via requestHelper.sendMaybeThrottle. Same shape as the
> sibling handleStreamsGroupHeartbeat. The coordinator-side method was
> added in KAFKA-20623 split 2 (#22552); only the KafkaApis side is
> missing — flagged on the split 2 review.
> Broker wiring (shared). BrokerServer.scala instantiates the plugin via
> config.getConfiguredInstance(...) and threads it into the
> GroupCoordinatorService builder (verify the describe path doesn't need
> additional wiring beyond what KAFKA-20623 already added).
> RequestConvertToJson adds entries for StreamsGroupTopologyDescriptionUpdate
> Request/Response and for the new topology-description fields on the
> describe response so request/response logging tools can render them.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
