Static membership in Kafka Connect

Viewed 431

I'm trying to implement the static membership in Kafka Connect. Our Kafka Connect cluster is deployed on K8S using the Strimzi Kafka operator.

I have tried putting the following config for the workers (in the KafkaConnect yaml):

connector.client.config.override.policy: All
consumer.group.instance.id: somethingsomething

And in the HttpSinkConnector class I have this:

@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
    List<Map<String, String>> configs = new ArrayList<>(maxTasks);
    for (int i = 0; i < maxTasks; i++) {
        Map<String, String> configCopy = new HashMap<>(this.configProps);
        configCopy.put("consumer.override.group.instance.id", Thread.currentThread().getName());

        configs.add(configCopy);
    }
    return configs;
}

This gave org.apache.kafka.common.errors.FencedInstanceIdException with some log - ...07:07:32,631 ERROR [Consumer instanceId=somethingsomething... because all tasks got somethingsomething as their group.instance.id although it should have get Thread.currentThread().getName().

I have also tried just the following (without the the worker configuration):

@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
    List<Map<String, String>> configs = new ArrayList<>(maxTasks);
    for (int i = 0; i < maxTasks; i++) {
        Map<String, String> configCopy = new HashMap<>(this.configProps);
        configCopy.put("consumer.group.instance.id", "somethingsomething");

        configs.add(configCopy);
    }
    return configs;
}

And that did nothing (no errors, no instanceId in logs) which mean I put this config value in the wrong place.

So how can I achieve static membership on Kafka Connect?

1 Answers

Use consumer.override.group.instance.id instead of consumer.group.instance.id

Related