...

Code Block

language	yml

/**
 * Prefix used to add arbitrary tags to a Kafka Stream's instance as key-value pairs.
 * Example:
 * client.tag.zone=zone1
 * client.tag.cluster=cluster1
 */
@SuppressWarnings("WeakerAccess")
public static final String CLIENT_TAG_PREFIX = "client.tag.";

We will also add a new configuration option in StreamsConfig, which will be the means of setting which tags Kafka Streams must take into account when balancing the standby tasks across the racks.



/**
 * Prefix a client tag key with {@link #CLIENT_TAG_PREFIX}.
 *
 * @param the client tag key
 * @return {@link #CLIENT_TAG_PREFIX} + {@code clientTagKey}
 */
pubic static String clientTagPrefix(final String clientTagKey) {
	return CLIENT_TAG_PREFIX + clientTagKey
}

We will also add a new configuration option in StreamsConfig, which will be the means of setting which tags Kafka Streams must take into account when balancing the standby tasks across the racks.

Code Block

language	yml

public static final String TASK_ASSIGNMENT_RACK_AWARENESS_CONFIG = "task.assignment.rack.awareness";
public static final String TASK_ASSIGNMENT_RACK_AWARENESS_DOC = "List of client tag keys used to distribute standby replicas across Kafka Streams instances." +

Code Block

language	yml

public static final String TASK_ASSIGNMENT_RACK_AWARENESS_CONFIG = "t
ask.assignment.rack.awareness";
public static final String TASK_ASSIGNMENT_RACK_AWARENESS_DOC = "List of client tag keys used to distribute standby replicas across Kafka Streams instances." +                                                                
                                           
                     " When configured, Kafka Streams will make a best effort to distribute" +
                    	           " When configured, Kafka Streams will make a best-effort to distribute" +
                    											" the standby tasks over each client tag dimension.";

...

Kafka Streams's Task Assignor will make a decision on how to distribute standby tasks over the available clients based on encoded clientTags within the subscription info and configured task.assignment.rack.awareness

Info

Standby task distribution algorithm is not specified in this KIP, but is left as an implementation detail. However, every distribution algorithm must handle gracefully when ideal standby task distribution is not possible; In that case, Kafka Streams must not fail the assignment but try to

find the subsequent most optimal distribution. The ideal distribution means there is no repeated client dimension amongst clients assigned to the active task and all standby tasks.

Changes in HighAvailabilityTaskAssignor

Implementation of this KIP must not affect HighAvailabilityTaskAssignor in a breaking way, meaning that all the existing behavior should stay unchanged (e.g., when new configurations are not specified). Once required configurations are set, the main change should happen within the code that deals with standby task allocation.

Compatibility, Deprecation, and Migration Plan

The changes proposed by this KIP shouldn't affect previously setup applications. Since we introduce new configuration options, existing ones shouldn't be affected by this change.

Rejected Alternatives

distribute the standby tasks on best-effort bases.

With an ideal task distribution, each client of the set of clients that host a given active task and the corresponding standby replicas has a unique value for each tag with regard to the other clients in the set.

Example of the ideal task distribution

Suppose we have the following infrastructure setup: Three Kubernetes Clusters, let us call them K8s_Cluster1, K8s_Cluster2, and K8s_Cluster3. Each Kubernetes cluster is spanned across three availability zones: eu-central-1a, eu-central-1b, eu-central-1c.

Our use-case is to have a distribution of the standby tasks across different Kubernetes clusters and AZs so we can be Kubernetes cluster and AZ failure tolerant.

With the new configuration options presented in this KIP, we will have the following:

Code Block

Node-1:
client.tag.cluster: K8s_Cluster1
client.tag.zone: eu-central-1a
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

Node-2:
client.tag.cluster: K8s_Cluster1
client.tag.zone: eu-central-1b
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

Node-3:
client.tag.cluster: K8s_Cluster1
client.tag.zone: eu-central-1c
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

Node-4:
client.tag.cluster: K8s_Cluster2
client.tag.zone: eu-central-1a
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

Node-5:
client.tag.cluster: K8s_Cluster2
client.tag.zone: eu-central-1b
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

Node-6:
client.tag.cluster: K8s_Cluster2
client.tag.zone: eu-central-1c
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

Node-7:
client.tag.cluster: K8s_Cluster3
client.tag.zone: eu-central-1a
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

Node-8:
client.tag.cluster: K8s_Cluster3
client.tag.zone: eu-central-1b
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

Node-9:
client.tag.cluster: K8s_Cluster3
client.tag.zone: eu-central-1c
task.assignment.rack.awareness: zone,cluster
num.standby.replicas: 2

With the infrastructure topology and configuration presented above, we can easily achieve The Ideal standby task distribution. The Ideal standby task distribution is achievable because we have to allocate three tasks for any given stateful task (1 active task + 2 standby task), and it corresponds to unique values for each tag.

Assuming active stateful task 0_0 is in Node-1, The Ideal standby task distribution might look like this:

Node-5 (different cluster, different zone), Node-9 (different cluster, different zone)
Node-6 (different cluster, different zone), Node-8 (different cluster, different zone)

Changes in HighAvailabilityTaskAssignor

Implementation of this KIP must not affect HighAvailabilityTaskAssignor in a breaking way, meaning that all the existing behavior should stay unchanged (e.g., when new configurations are not specified). Once required configurations are set, the main change should happen within the code that deals with standby task allocation.

Compatibility, Deprecation, and Migration Plan

The changes proposed by this KIP shouldn't affect previously setup applications. Since we introduce new configuration options, existing ones shouldn't be affected by this change.

Rejected Alternatives

The initial idea was to introduce two configurations in StreamsConfig, rack.id, which defines the rack of the Kafka Streams instance and standby.task.assignor - class that implements RackAwareStandbyTaskAssignor interface.

The signature of RackAwareStandbyTaskAssignor was the following:

Code Block

language	java

public interface RackAwareStandbyTaskAssignor {

    /**
     * Computes desired standby task distribution for a different {@link StreamsConfig#RACK_ID_CONFIG}s.
     * @param sourceTasks - Source {@link TaskId}s with a corresponding rack IDs that are eligible for standby task creation.
     * @param clientRackIds - Client rack IDs that were received during assignment.
     * @return - Map of the rack IDs to set of {@link TaskId}s. The return value can be used by {@link TaskAssignor}
     *           implementation to decide if the {@link TaskId} can be assigned to a client that is located in a given rack.
     */
    Map<String, Set<TaskId>> computeStandbyTaskDistribution(final Map<TaskId, String> sourceTasks,

The initial idea was to introduce two configurations in StreamsConfig, rack.id, which defines the rack of the Kafka Streams instance and standby.task.assignor - class that implements RackAwareStandbyTaskAssignor interface.

The signature of RackAwareStandbyTaskAssignor was the following:

Code Block

language	java

public interface RackAwareStandbyTaskAssignor {

    /**
     * Computes desired standby task distribution for a different {@link StreamsConfig#RACK_ID_CONFIG}s.
     * @param sourceTasks - Source {@link TaskId}s with a corresponding rack IDs that are eligible for standby task creation.
     * @param clientRackIds - Client rack IDs that were received during assignment.
     * @return - Map of the rack IDs to set of {@link TaskId}s. The return value can be used by {@link TaskAssignor}
     *           implementation to decide if the {@link TaskId} can be assigned to a client that is located in a given rack.
     */
    Map<String, Set<TaskId>> computeStandbyTaskDistribution(final Map<TaskId, String> sourceTasks,
                                                            final Set<String> clientRackIds);
}

By injecting custom implementation of RackAwareStandbyTaskAssignor interface, users could hint Kafka Streams where to allocate certain standby tasks when more complex processing logic was required — for example, parsing rack.id, which can be a combination of multiple identifiers (as seen in the previous examples where we have cluster and zone tags).

The above mentioned idea was abandoned because it's easier and more user-friendly to let users control standby task allocation with just configuration options instead of forcing them to implement a custom interface.

The second approach was to refactor TaskAssignor interface to be more user-friendly and expose it as a public interface. Users then could implement custom TaskAssignor logic and set it via StreamsConfig. With this, Kafka Streams users would effectively be in control of Active and Standby task allocation.
Similarly to the point above, this approach also was rejected because it's more complex.
Even though it's more-or-less agreed on the pluggable TaskAssignor interface's usefulness, it was decided to cut it out of this KIP's scope and prepare a separate one for that feature.

Client tags vs single rack.id configuration

Defining multiple client.tag with combination of task.assignment.rack.awareness gives more flexibility, which otherwise could have been only possible with pluggable custom logic Kafka Streams's user must provide (it is briefly described in "Rejected Alternatives" section).

For instance, if we append multiple tags to form a single rack, it may not give desired distribution to the user if the infrastructure topology is more complex. Let us consider the following example with appending multiple tags to form the single rack.

...

more user-friendly to let users control standby task allocation with just configuration options instead of forcing them to implement a custom interface.

Defining multiple client.tag with combination of task.assignment.rack.awareness gives more flexibility, which, as already mentioned above, could have been only possible with pluggable custom logic Kafka Streams's user must provide.

For instance, if we append multiple tags to form a single rack, it may not give desired distribution to the user if the infrastructure topology is more complex. Let us consider the following example with appending multiple tags to form the single rack.

Code Block

Node-1:
rack.id: K8s_Cluster1-eu-central-1a
num.standby.replicas: 1

Node-2:
rack.id: K8s_Cluster1-eu-central-1b
num.standby.replicas: 1

Node-3:
rack.id: K8s_Cluster1-eu-central-1c
num.standby.replicas: 1

Node-4:
rack.id: K8s_

...

Cluster2-eu-central-1a
num.standby.replicas: 1

Node-

...

5: rack.id: K8s_

...

Cluster2-eu-central-1b
num.standby.replicas: 1

Node-

...

6: rack.id: K8s_

...

Cluster2-eu-central-1c
num.standby.replicas: 1

...

In the example mentioned above, we have three AZs and two Kubernetes clusters. Our use-case is to distribute standby task in the different Kubernetes cluster and different availability zone. For instance, if the active task is in Node-1 (K8s_Cluster1-eu-central-1a

...

), the corresponding standby task should be in either on Node-5 (K8s_Cluster2-eu-central-1b

...

) or on Node-6 (K8s_Cluster2-eu-central-1c).
Unfortunately, without custom logic provided by the user, this would be very hard to achieve with a single rack.id configuration. Because without any input from the user, Kafka Streams might as well allocate standby task for the active task either:
- In the same Kubernetes cluster and different AZ (Node-2, Node-3)
- In the different Kubernetes cluster but the same AZ (Node-4)
On the other hand, with the combination of the new "client.tag.*" and "task.assignment.rack.awareness" configurations, standby task distribution algorithm will be able to figure out what will be the most optimal distribution by balancing the standby tasks over each client.tag dimension individually. And it can be achieved by simply providing necessary configurations to Kafka Streams.
The second approach was to refactor TaskAssignor interface to be more user-friendly and expose it as a public interface. Users then could implement custom TaskAssignor logic and set it via StreamsConfig. With this, Kafka Streams users would effectively be in control of Active and Standby task allocation.
Similarly to the point above, this approach also was rejected because it's more complex.
Even though it's more-or-less agreed on the pluggable TaskAssignor interface's usefulness, it was decided to cut it out of this KIP's scope and prepare a separate one for that feature

In the example mentioned above, we have three AZs and two Kubernetes clusters. Our use-case is to distribute standby task in the different Kubernetes cluster and different availability zone. For instance, if the active task is in Node-1 (K8s_Cluster1-eu-central-1a), the corresponding standby task should be in either on Node-5 (K8s_Cluster2-eu-central-1b) or on Node-6 (K8s_Cluster2-eu-central-1c).

Unfortunately, without custom logic provided by the user, this would be very hard to achieve with a single rack.id configuration. Because without any input from the user, Kafka Streams might as well allocate standby task for the active task either:

In the same Kubernetes cluster and different AZ (Node-2, Node-3)
In the different Kubernetes cluster but the same AZ (Node-4)

...

.

Space shortcuts

Child pages

Versions Compared

Old Version 28

New Version 29

Key

Changes in HighAvailabilityTaskAssignor

Compatibility, Deprecation, and Migration Plan

Rejected Alternatives

Example of the ideal task distribution

Changes in HighAvailabilityTaskAssignor

Compatibility, Deprecation, and Migration Plan

Rejected Alternatives

Client tags vs single rack.id configuration

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 28

New Version 29

Key

Changes in HighAvailabilityTaskAssignor

Compatibility, Deprecation, and Migration Plan

Rejected Alternatives

Example of the ideal task distribution

Changes in HighAvailabilityTaskAssignor

Compatibility, Deprecation, and Migration Plan

Rejected Alternatives

Client tags vs single rack.id configuration