set default clientId throws InstanceAlreadyExistsException

[PrabaharanK]

Hi Team

I'm using Greenwich.RC2 spring cloud version for my project. I try to set client.id specification for the kafka binder and gets below exception.

How to set the client.id property?, I tried to set the property on bindings level but no impact, I shared my application.yml as well.

javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=client1-0
	at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437) ~[na:1.8.0_161]
	at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898) ~[na:1.8.0_161]
	at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966) ~[na:1.8.0_161]
	at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900) ~[na:1.8.0_161]
	at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324) ~[na:1.8.0_161]
	at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) ~[na:1.8.0_161]
	at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:62) ~[kafka-clients-2.0.1.jar:na]
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:791) [kafka-clients-2.0.1.jar:na]
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:615) [kafka-clients-2.0.1.jar:na]
	at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:164) [spring-kafka-2.2.6.RELEASE.jar:2.2.6.RELEASE] 



spring:
  cloud:
    stream:
      binders:
        kafka1:
          type: kafka
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder:
                      brokers: localhost
                      autoCreateTopics: true
                      configuration:
                        default:
                          key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
                          value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
                        auto.offset.reset: latest
                        client.id: client1
      bindings:
        INPUT1:
          destination: INPUT_TOPIC
          group: INPUT_GROUP1
          binder: kafka1
        INPUT2:
          destination: INPUT_TOPIC2
          group: INPUT_GROUP1
          binder: kafka1
        OUTPUT1:
          destination: OUTPUT_TOPIC
          group: INPUT_GROUP1
          binder: kafka1

[garyrussell]

Why are you using an old release candidate? The current Greenwich version is SR6 https://spring.io/projects/spring-cloud#learn

You can't set the client id that way.

You can add a ListenerContainerCustomizer @Bean and

@Bean
public ListenerContainerCustomizer<AbstractKafkaListenerContainerFactory<?, ?>> customizer() {
    return (container, destinationName, group) -> container.getContainerProperties().setClientIdPrefix(...);

[PrabaharanK]

Hi Gary, Thank you, I updated the version to Greenwich SR6.

I tried the below code and no-luck, still the clientid field is empty. I verified the bean method is invoked during the startup.

   @Bean
    public ListenerContainerCustomizer<AbstractKafkaListenerContainerFactory<?, ?, ?>> customizer() {
        return (container, destinationName, group) -> container.getContainerProperties().setClientId("client1");
    }

In my console, below are the properties I can see.

2020-07-27 22:39:03.730  INFO 10612 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
	auto.commit.interval.ms = 100
	auto.offset.reset = latest
	bootstrap.servers = [localhost:9092]
	check.crcs = true
	client.id = 
	connections.max.idle.ms = 540000
	default.api.timeout.ms = 60000
	enable.auto.commit = false
	exclude.internal.topics = true
	fetch.max.bytes = 52428800
	fetch.max.wait.ms = 500
	fetch.min.bytes = 1
	group.id = INPUT_GROUP1
	heartbeat.interval.ms = 3000
	interceptor.classes = []
	internal.leave.group.on.close = true
	isolation.level = read_uncommitted
	key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
	max.partition.fetch.bytes = 1048576
	max.poll.interval.ms = 300000
	max.poll.records = 500
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	send.buffer.bytes = 131072
	session.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
	ssl.endpoint.identification.algorithm = https
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLS
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer

[garyrussell]

Sorry, I made a typo in the generic type; also you must set a unique prefix for each binding:

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
	return (container, destinationName, group) -> container.getContainerProperties().setClientId(destinationName + "." + group);
}

By the way, it is not good practice to put multiple input bindings in the same group - a rebalance on one consumer will force an unnecessary rebalance on the other(s).

[PrabaharanK]

Hi Gary, still clientId is empty.

`@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
	return (container, destinationName, group) -> container.getContainerProperties().setClientId(destinationName + "." + group);
}`

then I tried below code as well and I can see application started successfully and no trace of exception in console.

    @Bean
    public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
        return (container, destinationName, group) -> {
            if(true) throw new NullPointerException("Test");
            container.getContainerProperties().setClientId(destinationName + "." + group);
        };
    }

I'm using

        <dependency>
            <!-- Import dependency management from Spring Boot -->
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.1.5.Release</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>

[mborgraeve]

I'm having the same issue here, I'm trying to setup a client-id in order to be able to have clear logs when we startup the app and one cluster isn't available.

Currently the client-id of the adminclient beans are the generic ones, prefixes by adminclient and suffixed with -n with n being a number.
Such a log (adminclient-2):

2021-05-10 18:27:16.461  INFO 41669 --- [| adminclient-2] o.a.k.c.a.i.AdminMetadataManager         : [AdminClient clientId=adminclient-2] Metadata update failed

org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1620685636461) timed out at 9223372036854775807 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call.

I would like to change the client id so it's clear which cluster is down. This log can be reproduced with only one cluster (you can use nc -l -k 9092 to simulate a cluster accepting connections but timing out).

I can provide a minimal app if required (I'll work on it tomorrow).
I've tried the solution above without success. I also tried to set it up through several yaml properties without success.

We can set the value using the spring.kafka.admin.client-id property, but then it gets set on every admin for both clusters, so back to square one for me.

Note that if the cluster is responding, the consumer client ids are setup from the group value.

[mborgraeve]

Here is a project that can be used to show this specific case: https://github.com/mborgraeve/kafka-binder-client-id-sample

[omercelikceng]

Sorry, I made a typo in the generic type; also you must set a unique prefix for each binding:

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
	return (container, destinationName, group) -> container.getContainerProperties().setClientId(destinationName + "." + group);
}

By the way, it is not good practice to put multiple input bindings in the same group - a rebalance on one consumer will force an unnecessary rebalance on the other(s).

Sorry, I made a typo in the generic type; also you must set a unique prefix for each binding:

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
	return (container, destinationName, group) -> container.getContainerProperties().setClientId(destinationName + "." + group);
}

By the way, it is not good practice to put multiple input bindings in the same group - a rebalance on one consumer will force an unnecessary rebalance on the other(s).

I did not understand what you are saying here. What's the harm in giving the same groupId to different topics? Does it really cause rebalancing? Did I misunderstand? If so, can you explain why? When rebalancing, only the groupId is taken care of?@garyrussell

Your Answer : (By the way, it is not good practice to put multiple input bindings in the same group - a rebalance on one consumer will force an unnecessary rebalance on the other(s).)

[eomercelik]

Thank you @garyrussell . What you said worked. I was really able to customize client Ids. However, I have a small problem. I have 20 topics. I follow the logs. Different client Ids are assigned to the consumers of my 20 topics as I want. However, only 5 consumers have a problem afterwards. It prints the clientId log to them again and override the clientId that I assigned. New different clientIds are assigned for those 5 consumers. I don't understand why he behaved like this. I wonder if it takes my new clientId while doing rebalance. Why does this code work for some consumers and not others? Why is it suddenly assigned a different clientId? I'm doing something wrong but I don't know what.

[tooptoop4]

@PrabaharanK did u solve?