[jira] [Updated] (KAFKA-4117) Cleanup StreamPartitionAssignor behavior

["Guozhang Wang (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-4117?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Guozhang Wang updated KAFKA-4117:
---------------------------------
    Description: 
I went through the whole assignment logic once again and I feel the logic has now becomes a bit lossy, and I want to clean them up probably in another PR but just dump my thoughts here on the appropriate logic:

Some background:

1. Each {{KafkaStreams}} instance contains a clientId, and if not specified default value is applicationId-1/2/etc if there are multiple instances inside the same JVM. One instance contains multiple threads where the thread-clientId is constructed as clientId-StreamThread-1/2/etc, and the thread-clientId is used as the embedded consumer clientId as well as metrics tag.

2. However, since one instance can contain multiple threads, and hence multiple consumers, and when considering partition assignment, the streams library need to take the capacity into consideration based on the granularity of instance not on threads. Therefore we create a 4byte {{UUID.randomUUID()}} as the processId and encode that in the subscription metadata bytes, and the leader then knows if multiple consumer members are actually belong to the same instance (i.e. belong to threads of that instance), so that when assigning partitions it can balance among instances. NOTE that in production we recommend one thread per instance, so consumersByClient will only have one consumer per client (i.e. instance).

3. In addition, historically we hard-code the partition grouper logic, where for each task, it is assigned only with one partition of its subscribed topic. For example, if we have topicA with 5 partitions and topicB with 10 partitions, we will create 10 tasks, with the first five tasks containing one of the partitions each, while the last five tasks contain only one partition from topicB. And therefore the TaskId class contains the groupId of the sub-topology and the partition, so that taskId(group, 1) gets partition1 of topicA and partition1 of topicB. We later expose this to users to customize so that more than one partitions of the topic can be assigned to the same task, so that the partition field in the TaskId no longer indicate anything about which partitions are assigned, and we add {{AssignedPartitions}} to capture which partitions are assigned to which tasks.

4. While doing the assignment, the leader is also responsible for creating these changelog / repartition topics, and the number of partitions of these topics are equal to the number of tasks that needs to write to these topics, which are wrapped in {{stateChangelogTopicToTaskIds}} and {{internalSourceTopicToTaskIds}} respectively. After such topics are created, the leader also needs to "augment" the received cluster metadata with these topics to 1) check for copartitioning, and 2) maintained for QueryableState's discovery function.

The current implementation is mixed with all these legacy logic and gets quite messy, and I'm thinking to make a pass over the StreamPartitionAssignor and cleaning up it bit. More precisely:

1. Read and parse the subscription information to construct the clientMetadata map, where each metadata contains the {{Set<String> consumerMemberIds}}, {{ClientState<TaskId> state}}, and {{HostInfo hostInfo}}.

2. Access the (sub-)topology to create the corresponding changelog / repartition topics and construct the {{stateChangelogTopicToTaskIds}} and {{internalSourceTopicToTaskIds}}.

3. Call {{streamThread.partitionGrouper.partitionGroups}} to get the map from created tasks to their assigned partitions.

4. Call {{TaskAssignor.assign}} (which now takes the whole clientMetadata map) to assign tasks to clients, and hence we get the assigned partitions to clients.

5. For each client, use some round-robin manner (as we did now) to assign tasks to their hosted consumers with the {{clientMetadata.consumerMemberIds}} map.

6. Check co-partitioning of assigned partitions, and maintain the {{Cluster}} metadata locally on the leader.

7. Construct the assignment info, where activeTasks is also a map from {{TaskId}} to list of {{TopicPartitions}} since otherwise we will not know which partitions are assigned to which tasks.

8. For non-leaders, when getting the assignment, also construct the Cluster metadata from the decoded assignment information; and also maintain the AssignmentInfo locally for constructing the tasks.

And some minor improvements:

1. The default {{thread-clientIds applicationId-x-StreamThread-y}} may still be conflicting to each other with multiple JVMs / machines, which is bad for metrics collection / debugging across hosts. We can modify the default clientId to {{applicationId-processId}} whereprocessId is the generated UUID, hence the default thread-clientId is {{applicationId-UUID-StreamThread-y}}.

2. The {{TaskId.partition}} field no longer indicate which partitions are actually assigned to this task, but we still need to keep its topicGroupId field as it indicates which sub-topology this task belongs to, hence helpful for debugging. So maybe we can rename the partition field to sth. like sequence?

  was:
I went through the whole assignment logic once again and I feel the logic has now becomes a bit lossy, and I want to clean them up probably in another PR but just dump my thoughts here on the appropriate logic:

Some background:

1. Each {{KafkaStreams}} instance contains a clientId, and if not specified default value is applicationId-1/2/etc if there are multiple instances inside the same JVM. One instance contains multiple threads where the thread-clientId is constructed as clientId-StreamThread-1/2/etc, and the thread-clientId is used as the embedded consumer clientId as well as metrics tag.

2. However, since one instance can contain multiple threads, and hence multiple consumers, and when considering partition assignment, the streams library need to take the capacity into consideration based on the granularity of instance not on threads. Therefore we create a 4byte {{UUID.randomUUID()}} as the processId and encode that in the subscription metadata bytes, and the leader then knows if multiple consumer members are actually belong to the same instance (i.e. belong to threads of that instance), so that when assigning partitions it can balance among instances. NOTE that in production we recommend one thread per instance, so consumersByClient will only have one consumer per client (i.e. instance).

3. In addition, historically we hard-code the partition grouper logic, where for each task, it is assigned only with one partition of its subscribed topic. For example, if we have topicA with 5 partitions and topicB with 10 partitions, we will create 10 tasks, with the first five tasks containing one of the partitions each, while the last five tasks contain only one partition from topicB. And therefore the TaskId class contains the groupId of the sub-topology and the partition, so that taskId(group, 1) gets partition1 of topicA and partition1 of topicB. We later expose this to users to customize so that more than one partitions of the topic can be assigned to the same task, so that the partition field in the TaskId no longer indicate anything about which partitions are assigned, and we add {{AssignedPartitions}} to capture which partitions are assigned to which tasks.

4. While doing the assignment, the leader is also responsible for creating these changelog / repartition topics, and the number of partitions of these topics are equal to the number of tasks that needs to write to these topics, which are wrapped in {{stateChangelogTopicToTaskIds}} and 
{{internalSourceTopicToTaskIds}} respectively. After such topics are created, the leader also needs to "augment" the received cluster metadata with these topics to 1) check for copartitioning, and 2) maintained for QueryableState's discovery function.

The current implementation is mixed with all these legacy logic and gets quite messy, and I'm thinking to make a pass over the StreamPartitionAssignor and cleaning up it bit. More precisely:

1. Read and parse the subscription information to construct the clientMetadata map, where each metadata contains the {{Set<String> consumerMemberIds}}, {{ClientState<TaskId> state}}, and {{HostInfo hostInfo}}.

2. Access the (sub-)topology to create the corresponding changelog / repartition topics and construct the {{stateChangelogTopicToTaskIds}} and {{internalSourceTopicToTaskIds}}.

3. Call {{streamThread.partitionGrouper.partitionGroups}} to get the map from created tasks to their assigned partitions.

4. Call {{TaskAssignor.assign}} (which now takes the whole clientMetadata map) to assign tasks to clients, and hence we get the assigned partitions to clients.

5. For each client, use some round-robin manner (as we did now) to assign tasks to their hosted consumers with the {{clientMetadata.consumerMemberIds}} map.

6. Check co-partitioning of assigned partitions, and maintain the {{Cluster}} metadata locally on the leader.

7. Construct the assignment info, where activeTasks is also a map from {{TaskId}} to list of {{TopicPartitions}} since otherwise we will not know which partitions are assigned to which tasks.

8. For non-leaders, when getting the assignment, also construct the Cluster metadata from the decoded assignment information; and also maintain the AssignmentInfo locally for constructing the tasks.

And some minor improvements:

1. The default {{thread-clientIds applicationId-x-StreamThread-y}} may still be conflicting to each other with multiple JVMs / machines, which is bad for metrics collection / debugging across hosts. We can modify the default clientId to {{applicationId-processId}} whereprocessId is the generated UUID, hence the default thread-clientId is {{applicationId-UUID-StreamThread-y}}.

2. The {{TaskId.partition}} field no longer indicate which partitions are actually assigned to this task, but we still need to keep its topicGroupId field as it indicates which sub-topology this task belongs to, hence helpful for debugging. So maybe we can rename the partition field to sth. like sequence?


> Cleanup StreamPartitionAssignor behavior
> ----------------------------------------
>
>                 Key: KAFKA-4117
>                 URL: https://issues.apache.org/jira/browse/KAFKA-4117
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>            Reporter: Guozhang Wang
>            Assignee: Guozhang Wang
>              Labels: architecture
>
> I went through the whole assignment logic once again and I feel the logic has now becomes a bit lossy, and I want to clean them up probably in another PR but just dump my thoughts here on the appropriate logic:
> Some background:
> 1. Each {{KafkaStreams}} instance contains a clientId, and if not specified default value is applicationId-1/2/etc if there are multiple instances inside the same JVM. One instance contains multiple threads where the thread-clientId is constructed as clientId-StreamThread-1/2/etc, and the thread-clientId is used as the embedded consumer clientId as well as metrics tag.
> 2. However, since one instance can contain multiple threads, and hence multiple consumers, and when considering partition assignment, the streams library need to take the capacity into consideration based on the granularity of instance not on threads. Therefore we create a 4byte {{UUID.randomUUID()}} as the processId and encode that in the subscription metadata bytes, and the leader then knows if multiple consumer members are actually belong to the same instance (i.e. belong to threads of that instance), so that when assigning partitions it can balance among instances. NOTE that in production we recommend one thread per instance, so consumersByClient will only have one consumer per client (i.e. instance).
> 3. In addition, historically we hard-code the partition grouper logic, where for each task, it is assigned only with one partition of its subscribed topic. For example, if we have topicA with 5 partitions and topicB with 10 partitions, we will create 10 tasks, with the first five tasks containing one of the partitions each, while the last five tasks contain only one partition from topicB. And therefore the TaskId class contains the groupId of the sub-topology and the partition, so that taskId(group, 1) gets partition1 of topicA and partition1 of topicB. We later expose this to users to customize so that more than one partitions of the topic can be assigned to the same task, so that the partition field in the TaskId no longer indicate anything about which partitions are assigned, and we add {{AssignedPartitions}} to capture which partitions are assigned to which tasks.
> 4. While doing the assignment, the leader is also responsible for creating these changelog / repartition topics, and the number of partitions of these topics are equal to the number of tasks that needs to write to these topics, which are wrapped in {{stateChangelogTopicToTaskIds}} and {{internalSourceTopicToTaskIds}} respectively. After such topics are created, the leader also needs to "augment" the received cluster metadata with these topics to 1) check for copartitioning, and 2) maintained for QueryableState's discovery function.
> The current implementation is mixed with all these legacy logic and gets quite messy, and I'm thinking to make a pass over the StreamPartitionAssignor and cleaning up it bit. More precisely:
> 1. Read and parse the subscription information to construct the clientMetadata map, where each metadata contains the {{Set<String> consumerMemberIds}}, {{ClientState<TaskId> state}}, and {{HostInfo hostInfo}}.
> 2. Access the (sub-)topology to create the corresponding changelog / repartition topics and construct the {{stateChangelogTopicToTaskIds}} and {{internalSourceTopicToTaskIds}}.
> 3. Call {{streamThread.partitionGrouper.partitionGroups}} to get the map from created tasks to their assigned partitions.
> 4. Call {{TaskAssignor.assign}} (which now takes the whole clientMetadata map) to assign tasks to clients, and hence we get the assigned partitions to clients.
> 5. For each client, use some round-robin manner (as we did now) to assign tasks to their hosted consumers with the {{clientMetadata.consumerMemberIds}} map.
> 6. Check co-partitioning of assigned partitions, and maintain the {{Cluster}} metadata locally on the leader.
> 7. Construct the assignment info, where activeTasks is also a map from {{TaskId}} to list of {{TopicPartitions}} since otherwise we will not know which partitions are assigned to which tasks.
> 8. For non-leaders, when getting the assignment, also construct the Cluster metadata from the decoded assignment information; and also maintain the AssignmentInfo locally for constructing the tasks.
> And some minor improvements:
> 1. The default {{thread-clientIds applicationId-x-StreamThread-y}} may still be conflicting to each other with multiple JVMs / machines, which is bad for metrics collection / debugging across hosts. We can modify the default clientId to {{applicationId-processId}} whereprocessId is the generated UUID, hence the default thread-clientId is {{applicationId-UUID-StreamThread-y}}.
> 2. The {{TaskId.partition}} field no longer indicate which partitions are actually assigned to this task, but we still need to keep its topicGroupId field as it indicates which sub-topology this task belongs to, hence helpful for debugging. So maybe we can rename the partition field to sth. like sequence?



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)