Kafka consumer group

This pattern provides:

• Continual processing of messages from all partitions in the event of failure
• No assumption about which partition is consumed by which consumer operator, thus no guarantee that a message with key 'K' will be processed by the same operator.
• When partitions are added to the subscribed topic, these new partitions will be automatically assigned to one of the consumers in the group.

• automatically assign operators to partitions
• reassign partitions during a failure

For example, suppose there are six partitions on a topic. The following will occur for the different number of consumers:

• Three consumer operators
  • Each will pick two partitions (1/3 of the messages assuming even load distribution over partitions) during normal operation
  • When one operator is stopped or down for a longer period of time, the six partitions are distributed across the two remaining operators, each with three
  • Once the failed operator restarts, the partitions will again be redistributed across the three operators
• Six consumer operators
  • Each operator will pick one partition during normal operation, processing 1/6 of message volume
  • On failure of one operator, one of the remaining five will take over the partition, one of the operators will consume two partitions, the other four one.
• Seven consumer operators
  • Six operators will pick up a partition each, one operator will be idle.
  • On failure of one operator, the idle operator takes the partition of the failed one.

Partition de-assignment and re-assignment can happen when:

• Group management related timers expire (for example, the heart-beat timeout session.timeout.ms or the poll timeout max.poll.interval.ms).
• The group coordinator node goes down
• Meta data of the subscribed topic changes (for example, the number of partitions)

• Pro: High volume by having multiple operators reading messages in parallel from partitions
• Pro: Takeover of partitions from failed or stopped consumers by other members of the consumer group.
• Pro: No manual assignment of partitions, any number of operators will always correctly read all messages.
• Con: Keyed messages may be handled by any operator after failure and reassignment. As a workaround, the messages can be repartitioned by the message key in the Streams application with abutting parallel region.

• Consistent region: Supported (periodic only)
• Checkpointing via config checkpoint: Supported, but ignored. The operator does not save any data.

• No assignment of partitions is configured through the partition operator parameter.
• A group identifier must be specified either by the consumer property group.id, or by using the groupId parameter, which would have precedence over a bare property. All operators of the consumer group must share this group identifier.
• When not in a consistent region, the startPosition parameter must not be specified or must have the value Default, Beginning, End, or Time.
• The startPosition parameter must not be Offset.

• Create a composite containing the KafkaConsumer invocation if you want to add more operators to the parallel channel besides the KafkaConsumer
• Annotate KafkaConsumer or the composite invocation with @parallel with width N (e.g. width = 3 to handle 6 partitions).

• Invoke N copies of the KafkaConsumer operator.

Without consistent region

public composite ConsumerGroup {
param
    expression <int32> $N: (int32) getSubmissionTimeValue ("consumerGroupSize", "3");
graph
    @parallel (width = $N)
    stream <rstring json, rstring messageKey> Messages = KafkaConsumer() {
        param
            propertiesFile: "etc/consumer.properties";
            topic: "myTopic";
            groupId: "myConsumerGroup";
            outputMessageAttributeName: "json";
            outputKeyAttributeName: "messageKey";
            commitPeriopd: 10.0;      // commit offsets every 10 seconds
        config placement: partitionExlocation ("A");
    }

    // do partitioned processing in Streams
    // messages with same key go always into the same parallel channel
    @parallel (width = 4, partitionBy = [{port = Messages, attributes = [messageKey]}])
    stream <rstring json> Processed = Processing (Messages) {
    }

    ...
}

public composite Processing (input In; output Out) {
graph
    ...
}

public composite ConsumerGroupCR {
param
    expression <int32> $N: (int32) getSubmissionTimeValue ("consumerGroupSize", "3");
graph
    () as JCP = JobControlPlane() {}

    @consistent (trigger = periodic, period = 60.0, drainTimeout = 300.0, maxConsecutiveResetAttempts = 10)
    @parallel (width = $N)
    stream <rstring json, rstring messageKey> Messages = KafkaConsumer() {
        param
            propertiesFile: "etc/consumer.properties";
            topic: "myTopic";
            outputMessageAttributeName: "json";
            outputKeyAttributeName: "messageKey";
        config placement: partitionExlocation ("A");
    }

    // do partitioned processing in Streams
    // messages with same key go always into the same parallel channel
    @parallel (width = 4, partitionBy = [{port = Messages, attributes = [messageKey]}])
    stream <rstring json> Processed = Processing (Messages) {
    }
    ...
}

public composite Processing (input In; output Out) {
graph
    ...
}

group.id=myConsumerGroup