[
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, 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.
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.
Metrics (KIP-1331). Adds streams-group-topology-description-get-
{success, error}
-\{rate,count} sensors on GroupCoordinatorMetrics, recorded in the
manager's attachTopologyDescriptions building block — mirrors the
delete-\{success,error} sensors filled in by KAFKA-20696. KIP-1331 also spec'd
set-\{success,error}-\{rate,count} for plugin.setTopology that KAFKA-20623
shipped without; those touch the write path and are out of scope for this
ticket (a separate small follow-up should track 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, 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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
