SASL Authentication Error with Separate Clusters and Access Levels #2937

hemantmaersk · 2023-12-14T14:48:07Z

hemantmaersk
Dec 14, 2023

Description:

Issue:
I am encountering a SASL authentication error when attempting to use Kafka with separate clusters for producers and consumers, each requiring different access levels. User reading from a cluster has read only access. It is important to note that I am able to produce and consume successfully with observability turned on, when I have different topics in same cluster with user having read and write access and configured in admin. I tried a solution by configuring different admin for consumer and producer under but it did not work.

The error message is as follows:

Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512.

Configuration details for Kafka admin, producer and consumer.

spring:
    kafka:
        admin:
        client-id: myorg-client
        properties:
            security.protocol: SASL_SSL
            client.id: myorg-client
            sasl.mechanism: SCRAM-SHA-512
            sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="${KAFKA_PRODUCER_USERNAME}" password="${KAFKA_PRODUCER_PASSWORD}";
            ssl.truststore.location: ${KAFKA_PRODUCER_SSL_TRUSTSTORE_LOCATION}
            ssl.truststore.password: ${KAFKA_PRODUCER_SSL_TRUSTSTORE_PASSWORD}
            ssl.endpoint.identification.algorithm: https
        producer:
            bootstrap-servers: ${KAFKA_PRODUCER_BOOTSTRAP_SERVERS}
            key-serializer: org.apache.kafka.common.serialization.StringSerializer
            value-serializer: org.apache.kafka.common.serialization.StringSerializer
            properties:
                security.protocol: SASL_SSL
                client.id: myorg-client
                sasl.mechanism: SCRAM-SHA-512
                sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="${KAFKA_PRODUCER_USERNAME}" password="${KAFKA_PRODUCER_PASSWORD}";
                ssl.truststore.location: ${KAFKA_PRODUCER_SSL_TRUSTSTORE_LOCATION}
                ssl.truststore.password: ${KAFKA_PRODUCER_SSL_TRUSTSTORE_PASSWORD}
                ssl.endpoint.identification.algorithm: https
        consumer:
            bootstrap-servers: ${KAFKA_CONSUMER_BOOTSTRAP_SERVERS}
            key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
            value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
            enable-auto-commit: false
            auto-offset-reset: earliest
            properties:
                # Delegate deserializers
                spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
                spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
                security.protocol: SASL_SSL
                sasl.mechanism: SCRAM-SHA-512
                sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="${KAFKA_CONSUMER_USERNAME}" password="${KAFKA_CONSUMER_PASSWORD}";
                ssl.truststore.location: ${KAFKA_CONSUMER_SSL_TRUSTSTORE_LOCATION}
                ssl.truststore.password: ${KAFKA_CONSUMER_SSL_TRUSTSTORE_PASSWORD}
                ssl.endpoint.identification.algorithm: https
        listener:
             ack-mode: manual
            observation-enabled: true
        template:
            observation-enabled: true

Environment:

Spring Boot version: 3.2.0
Kafka version: 3.4.0

Steps to Reproduce:

Configure Kafka with separate clusters for producers and consumers (Cluster A - Read, Cluster B - Write).
Create a service which reads from Cluster A with user that has read only access.
Set up consumers configuration in service using spring.kafka.consumer with read access to a topic in a different cluster.
Configure producers using spring.kafka.producer with write access to a topic in a different cluster.
Configure admin using spring.kafka.admin with user of Cluster A that has read only access.
Observe the SASL authentication error during authentication.

Expected Behavior:
When I enabled observability, spring.kafka.listener.observation-enabled: true and spring.kafka.template.observation-enabled: true I expect to be able to read from a topic in a different cluster using a consumer with read-only access and write to a topic in a different cluster using a producer with write access.

Actual Behavior:
I am facing a SASL authentication error, preventing successful authentication during the setup described above.

Additional Information:

Logs

2023-12-14T14:42:05+05:30	09:12:05.728 [org.springframework.kafka.KafkaListenerEndpointContainer#3-0-C-1] DEBUG [my-app, traceId: 657ac6e53afb245eaf7ae98e7ad653c6, spanId: af7ae98e7ad653c6] --- Received message from topic: myorg.abc.test.topic.internal.any.v1
2023-12-14T14:42:05+05:30	Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512
2023-12-14T14:42:05+05:30		at java.base/java.lang.Thread.run(Thread.java:831)
2023-12-14T14:42:05+05:30		at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1322)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1451)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1487)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2132)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2490)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:2604)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2746)
2023-12-14T14:42:05+05:30		at io.micrometer.observation.docs.ObservationDocumentation.observation(ObservationDocumentation.java:188)
2023-12-14T14:42:05+05:30		at io.micrometer.observation.Observation.createNotStarted(Observation.java:172)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.lambda$doInvokeRecordListener$55(KafkaMessageListenerContainer.java:2749)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.support.micrometer.KafkaRecordReceiverContext.<init>(KafkaRecordReceiverContext.java:51)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.clusterId(KafkaMessageListenerContainer.java:1009)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.obtainClusterId(KafkaMessageListenerContainer.java:1016)
2023-12-14T14:42:05+05:30		at org.springframework.kafka.core.KafkaAdmin.clusterId(KafkaAdmin.java:335)
2023-12-14T14:42:05+05:30		at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:180)
2023-12-14T14:42:05+05:30		at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2096)
2023-12-14T14:42:05+05:30		at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
2023-12-14T14:42:05+05:30	java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512
2023-12-14T14:42:05+05:30	09:12:05.728 [org.springframework.kafka.KafkaListenerEndpointContainer#3-0-C-1] ERROR [my-app, traceId: , spanId: ] --- Could not obtain cluster info
2023-12-14T14:42:05+05:30	09:12:05.728 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- Metrics reporters closed
2023-12-14T14:42:05+05:30	09:12:05.728 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- Closing reporter org.apache.kafka.common.metrics.JmxReporter
2023-12-14T14:42:05+05:30	09:12:05.728 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- Metrics scheduler closed
2023-12-14T14:42:05+05:30	09:12:05.727 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- [AdminClient clientId=my-app] Timed out 2 remaining operation(s) during close.
2023-12-14T14:42:05+05:30	org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited. Call: fetchMetadata
2023-12-14T14:42:05+05:30	09:12:05.727 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- [AdminClient clientId=my-app] Metadata update failed
2023-12-14T14:42:05+05:30	org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited. Call: fetchMetadata
2023-12-14T14:42:05+05:30	09:12:05.727 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- [AdminClient clientId=my-app] Metadata update failed
2023-12-14T14:42:05+05:30	09:12:05.727 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- App info kafka.admin.client for my-app unregistered
2023-12-14T14:42:05+05:30	org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512
2023-12-14T14:42:05+05:30	09:12:05.726 [kafka-admin-client-thread | my-app] WARN  [my-app, traceId: , spanId: ] --- [AdminClient clientId=my-app] Metadata update failed due to authentication error
2023-12-14T14:42:05+05:30	09:12:05.726 [kafka-admin-client-thread | my-app] ERROR [my-app, traceId: , spanId: ] --- [AdminClient clientId=my-app] Connection to node -1 (emp-kafka-shared-ppwe-retina.my-org-digital.net/10.11.97.26:443) failed authentication due to: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512
2023-12-14T14:42:05+05:30	09:12:05.726 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- [AdminClient clientId=my-app] Node -1 disconnected.
2023-12-14T14:42:05+05:30	09:12:05.726 [kafka-admin-client-thread | my-app] INFO  [my-app, traceId: , spanId: ] --- [AdminClient clientId=my-app] Failed authentication with emp-kafka-shared-ppwe-retina.my-org-digital.net/10.11.97.26 (channelId=-1) (Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512)
2023-12-14T14:42:05+05:30	09:12:05.589 [org.springframework.kafka.KafkaListenerEndpointContainer#3-0-C-1] INFO  [my-app, traceId: , spanId: ] --- Kafka startTimeMs: 1702545125589
2023-12-14T14:42:05+05:30	09:12:05.589 [org.springframework.kafka.KafkaListenerEndpointContainer#3-0-C-1] INFO  [my-app, traceId: , spanId: ] --- Kafka commitId: 60e845626d8a465a
2023-12-14T14:42:05+05:30	09:12:05.589 [org.springframework.kafka.KafkaListenerEndpointContainer#3-0-C-1] INFO  [my-app, traceId: , spanId: ] --- Kafka version: 3.6.0
2023-12-14T14:42:05+05:30	
2023-12-14T14:42:05+05:30		ssl.truststore.type = JKS
2023-12-14T14:42:05+05:30		ssl.truststore.password = [hidden]
2023-12-14T14:42:05+05:30		ssl.truststore.location = /my-orgams/ca.p12
2023-12-14T14:42:05+05:30		ssl.truststore.certificates = null
2023-12-14T14:42:05+05:30		ssl.trustmanager.algorithm = PKIX
2023-12-14T14:42:05+05:30		ssl.secure.random.implementation = null
2023-12-14T14:42:05+05:30		ssl.provider = null
2023-12-14T14:42:05+05:30		ssl.protocol = TLSv1.3
2023-12-14T14:42:05+05:30		ssl.keystore.type = JKS
2023-12-14T14:42:05+05:30		ssl.keystore.password = null
2023-12-14T14:42:05+05:30		ssl.keystore.location = null
2023-12-14T14:42:05+05:30		ssl.keystore.key = null
2023-12-14T14:42:05+05:30		ssl.keystore.certificate.chain = null
2023-12-14T14:42:05+05:30		ssl.keymanager.algorithm = SunX509
2023-12-14T14:42:05+05:30		ssl.key.password = null
2023-12-14T14:42:05+05:30		ssl.engine.factory.class = null
2023-12-14T14:42:05+05:30		ssl.endpoint.identification.algorithm = https
2023-12-14T14:42:05+05:30		ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
2023-12-14T14:42:05+05:30		ssl.cipher.suites = null
2023-12-14T14:42:05+05:30		socket.connection.setup.timeout.ms = 10000
2023-12-14T14:42:05+05:30		socket.connection.setup.timeout.max.ms = 30000
2023-12-14T14:42:05+05:30		send.buffer.bytes = 131072
2023-12-14T14:42:05+05:30		security.providers = null
2023-12-14T14:42:05+05:30		security.protocol = SASL_SSL
2023-12-14T14:42:05+05:30		sasl.oauthbearer.token.endpoint.url = null
2023-12-14T14:42:05+05:30		sasl.oauthbearer.sub.claim.name = sub
2023-12-14T14:42:05+05:30		sasl.oauthbearer.scope.claim.name = scope
2023-12-14T14:42:05+05:30		sasl.oauthbearer.jwks.endpoint.url = null
2023-12-14T14:42:05+05:30		sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
2023-12-14T14:42:05+05:30		sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
2023-12-14T14:42:05+05:30		sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
2023-12-14T14:42:05+05:30		sasl.oauthbearer.expected.issuer = null
2023-12-14T14:42:05+05:30		sasl.oauthbearer.expected.audience = null
2023-12-14T14:42:05+05:30		sasl.oauthbearer.clock.skew.seconds = 30
2023-12-14T14:42:05+05:30		sasl.mechanism = SCRAM-SHA-512
2023-12-14T14:42:05+05:30		sasl.login.retry.backoff.ms = 100
2023-12-14T14:42:05+05:30		sasl.login.retry.backoff.max.ms = 10000
2023-12-14T14:42:05+05:30		sasl.login.refresh.window.jitter = 0.05
2023-12-14T14:42:05+05:30		sasl.login.refresh.window.factor = 0.8
2023-12-14T14:42:05+05:30		sasl.login.refresh.min.period.seconds = 60
2023-12-14T14:42:05+05:30		sasl.login.refresh.buffer.seconds = 300
2023-12-14T14:42:05+05:30		sasl.login.read.timeout.ms = null
2023-12-14T14:42:05+05:30		sasl.login.connect.timeout.ms = null
2023-12-14T14:42:05+05:30		sasl.login.class = null
2023-12-14T14:42:05+05:30		sasl.login.callback.handler.class = null
2023-12-14T14:42:05+05:30		sasl.kerberos.ticket.renew.window.factor = 0.8
2023-12-14T14:42:05+05:30		sasl.kerberos.ticket.renew.jitter = 0.05
2023-12-14T14:42:05+05:30		sasl.kerberos.service.name = null
2023-12-14T14:42:05+05:30		sasl.kerberos.min.time.before.relogin = 60000
2023-12-14T14:42:05+05:30		sasl.kerberos.kinit.cmd = /usr/bin/kinit
2023-12-14T14:42:05+05:30		sasl.jaas.config = [hidden]
2023-12-14T14:42:05+05:30		sasl.client.callback.handler.class = null
2023-12-14T14:42:05+05:30		retry.backoff.ms = 100
2023-12-14T14:42:05+05:30		retries = 2147483647
2023-12-14T14:42:05+05:30		request.timeout.ms = 30000
2023-12-14T14:42:05+05:30		reconnect.backoff.ms = 50
2023-12-14T14:42:05+05:30		reconnect.backoff.max.ms = 1000
2023-12-14T14:42:05+05:30		receive.buffer.bytes = 65536
2023-12-14T14:42:05+05:30		metrics.sample.window.ms = 30000
2023-12-14T14:42:05+05:30		metrics.recording.level = INFO
2023-12-14T14:42:05+05:30		metrics.num.samples = 2
2023-12-14T14:42:05+05:30		metric.reporters = []
2023-12-14T14:42:05+05:30		metadata.max.age.ms = 300000
2023-12-14T14:42:05+05:30		default.api.timeout.ms = 60000
2023-12-14T14:42:05+05:30		connections.max.idle.ms = 300000
2023-12-14T14:42:05+05:30		client.id = my-app
2023-12-14T14:42:05+05:30		client.dns.lookup = use_all_dns_ips
2023-12-14T14:42:05+05:30		bootstrap.servers = [emp-kafka-shared-ppwe-retina.my-org-digital.net:443]
2023-12-14T14:42:05+05:30		auto.include.jmx.reporter = true

Answered by artembilan

Dec 15, 2023

The kafkaListenerContainerFactory bean in the Spring Boot auto-configuration expects this:

ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer

I see that you do this ContainerCustomizer<String, GenericRecord, ConcurrentMessageListenerContainer<String, GenericRecord>>

It feels like Spring cannot inject your customizer because its generics don't fit into an expectations.
Consider to change its to the same same Object-based.

View full answer

artembilan · 2023-12-14T15:08:47Z

artembilan
Dec 14, 2023
Maintainer

It looks like your user does not have permissions to obtain cluster info from admin:

	public String clusterId() {
		if (this.clusterId == null) {
			try (AdminClient client = createAdmin()) {
				this.clusterId = client.describeCluster().clusterId().get(this.operationTimeout, TimeUnit.SECONDS);
				if (this.clusterId == null) {
					this.clusterId = "null";
				}
			}
			catch (InterruptedException ex) {
				Thread.currentThread().interrupt();
			}
			catch (Exception ex) {
				LOGGER.error(ex, "Could not obtain cluster info");
			}
		}
		return this.clusterId;
	}

Starting with version 3.1, the KafkaAdmin has this new API:

	/**
	 * Set the cluster id. Use this to prevent attempting to fetch the cluster id
	 * from the broker, perhaps if the user does not have admin permissions.
	 * @param clusterId the clusterId to set
	 * @since 3.1
	 */
	public void setClusterId(String clusterId) {

See similar discussion: #2923

11 replies

hemantmaersk Dec 14, 2023
Author

HI @artembilan

I've grasped the idea that I should configure a new Kafka Admin in the Kafka template using the setClusterId method (Below is the code). However, if you please pardon my naïveté , could you kindly provide some example code or share a reference link that demonstrates how to inject the cluster ID into the AbstractMessageListenerContainer so that it will be accessible to logs while tracing in @KafkaListener?
Additionally, could you please review the code where I access the Kafka Template created by the Spring Framework and integrate a new KafkaAdmin instance with the cluster ID?

  @Bean
  public KafkaTemplate<String, String> customProducerKafkaTemplate(KafkaTemplate<String, String> kafkaTemplate,
      KafkaAdmin customProducerKafkaAdmin) {
    // Clone existing KafkaTemplate properties
    KafkaTemplate<String, String> customKafkaTemplate = new KafkaTemplate<>(kafkaTemplate.getProducerFactory());
    customKafkaTemplate.setDefaultTopic(kafkaTemplate.getDefaultTopic());
    customKafkaTemplate.setKafkaAdmin(customProducerKafkaAdmin);

    // You can add more customization to the KafkaTemplate here

    return customKafkaTemplate;
  }

  @Bean
  public KafkaAdmin customProducerKafkaAdmin() {
    // Create a new KafkaAdmin for the producer with different configurations 
    Map<String, Object> adminConfigs = new HashMap<>();

    KafkaAdmin kafkaAdmin = new KafkaAdmin(adminConfigs);
    
    //Just set cluster id and no other properties
    kafkaAdmin.setClusterId("Your Cluster ID");
    return kafkaAdmin;
  }

artembilan Dec 14, 2023
Maintainer

So, I would go an auto-configuration for the KafkaTemplate and KafkaAdmin for producer side.
And then would do this:

@Bean 
String kafkaTemplateAdminConfigurer(KafkaAdmin kafkaAdmin) {
    kafkaAdmin.setClusterId("Your Cluster ID");
    return null;
}

For the consumer side there is a ContainerCustomizer abstraction:

@Bean 
ContainerCustomizer<?, ?, ConcurrentMessageListenerContainer> consumerCustomizer(KafkaProperties properties) {
     return (container) -> {
         container.setKafkaAdmin();
     };
}

And to avoid confusion with an auto-configuration, I would advice to create this container-specific KafkaAdmin manually just here in this customizer without creating respective bean.
You still can borrow consumer configs from the mentioned KafkaProperties to populate options for that KafkaAdmin.

hemantmaersk Dec 15, 2023
Author

Hi @artembilan : Thank you so much for the help, Kafka Template is working fine even with Autoconfiguration as it has a write access but Listener is not working. I tried to use above code for consumer Customizer, it is invoked but admin is not getting updated. Is there a way to access KafkaMessageListenerContainer so that I can set admin and cluster id directly in that? Tried to use following code but bean was not initialised probably because it is Async processing

public class ApplicationInitializer implements ApplicationListener<ApplicationEvent>

Also can I log request in Spring boot to have cluster id property set for both producer and consumer via config properties so that admin access is not needed?

It would be really great if you could help with consumer code to set cluster id as above one did not work.

  @Bean
  ContainerCustomizer<String, GenericRecord, ConcurrentMessageListenerContainer<String,
        GenericRecord>> consumerCustomizer() {
    Map<String, Object> configs = new HashMap<>();
    configs.put("security.protocol", "emp-kafka-myorg-digital.net:443");
    configs.put("clusterId", "Cluster ID");
    return (container) -> {
      KafkaAdmin kafkaAdmin = new KafkaAdmin(configs);
      kafkaAdmin.setClusterId("Cluster ID");
      container.setKafkaAdmin(kafkaAdmin);
    };
  }

artembilan Dec 15, 2023
Maintainer

The kafkaListenerContainerFactory bean in the Spring Boot auto-configuration expects this:

ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer

I see that you do this ContainerCustomizer<String, GenericRecord, ConcurrentMessageListenerContainer<String, GenericRecord>>

It feels like Spring cannot inject your customizer because its generics don't fit into an expectations.
Consider to change its to the same same Object-based.

Answer selected by hemantmaersk

artembilan Dec 15, 2023
Maintainer

Also, I don't see that you provide any brokers info into that KafkaAdmin instance you create on the fly.

hemantmaersk Dec 18, 2023
Author

Thanks a ton @artembilan , Really appreciate your help and support! It worked!

You are a life saver!

hemantmaersk Jan 17, 2024
Author

Hi @artembilan :

Following to our discussion above, we have a use case where we need to connect to different Kafka Brokers of different applications to get data. We create consumers as follows for different applications by configuring properties in yml.

how to inject different ContainerCustomizer for different consumers with cluster id for different Kafka Admin?

ContainerCustomizer<String, GenericRecord, ConcurrentMessageListenerContainer<String, GenericRecord>>
Your help in this will be greatly appreciated!

Code to create consumer:


@Configuration
@ConfigurationProperties(“my-app.kafka.consumer")
public class KafkaConsumerConfig extends KafkaProperties.Consumer {

  @Bean
  public ConsumerFactory<String, String> appConsumerFactory() {
    return new DefaultKafkaConsumerFactory<>(buildProperties(null));
  }

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, String> appKafkaContainerFactory(
      ConsumerFactory<String, String> appConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory
        = new ConcurrentKafkaListenerContainerFactory<>();
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
    factory.setConsumerFactory(consumerFactory);
    return factory;
  }
}

artembilan Jan 17, 2024
Maintainer

@hemantmaersk ,

sorry, but your last question doesn't look like belongs to this conversation.
Please, consider to ask a new one on StackOverflow (yes, I prefer there, since more community support).

In general: you cannot do that with Spring Boot. When you do your own config, the ContainerCustomizer is out of the picture and you simply configure that ConcurrentKafkaListenerContainerFactory whatever way you need.

SASL Authentication Error with Separate Clusters and Access Levels #2937

Uh oh!

Uh oh!

hemantmaersk Dec 14, 2023

Replies: 1 comment · 11 replies

Uh oh!

artembilan Dec 14, 2023 Maintainer

Uh oh!

Uh oh!

hemantmaersk Dec 14, 2023 Author

Uh oh!

artembilan Dec 14, 2023 Maintainer

Uh oh!

hemantmaersk Dec 15, 2023 Author

Uh oh!

artembilan Dec 15, 2023 Maintainer

Uh oh!

artembilan Dec 15, 2023 Maintainer

Uh oh!

Uh oh!

hemantmaersk Dec 18, 2023 Author

Uh oh!

hemantmaersk Jan 17, 2024 Author

Uh oh!

artembilan Jan 17, 2024 Maintainer

hemantmaersk
Dec 14, 2023

Replies: 1 comment 11 replies

artembilan
Dec 14, 2023
Maintainer

hemantmaersk Dec 14, 2023
Author

artembilan Dec 14, 2023
Maintainer

hemantmaersk Dec 15, 2023
Author

artembilan Dec 15, 2023
Maintainer

artembilan Dec 15, 2023
Maintainer

hemantmaersk Dec 18, 2023
Author

hemantmaersk Jan 17, 2024
Author

artembilan Jan 17, 2024
Maintainer