Topic authorization failed for topics

[vkpuduran]

Hi, I have multiple KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> to consume messages from different topics. I am creating a factory for each topic. My code is as below,

@Bean(name = "xxxEventKafkaListenerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> xxxAlertEventKafkaListenerFactory() throws AWSServiceProviderException {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactoryUsingTopicName(ps-mon.0));
factory.setBatchListener(true);
factory.setConcurrency(numberOfConcurrentListeners);
factory.getContainerProperties().setAckMode(AckMode.valueOf(listenerAckMode));
    
 return factory;
}

consumerFactoryUsingTopicName(String topicName) -> returns the default config values like,

    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, maxPartitionFetchBytes);
    props.put(CONST_SASL_MECHANISM, SCRAM_SHA_512); 
    props.put(CONST_SASL_JAAS_CONFIG, jaasConfig); 
    props.put(CONST_SECURITY_PROTOCOL, SASL_PROTOCOL);

Few of the KafkaListenerContainerFactories use the same SASL credentials (user name and pwd) - jaasConfig -> org.apache.kafka.common.security.scram.ScramLoginModule required username="%s" password="%s";
Whereas the other KafkaListenerContainerFactory uses different SASL credentials (user name and pwd). This is causing some issue and I am getting the below error.

**
2021-05-07 19:31:52 ERROR o.a.k.c.Metadata:283 [org.springframework.kafka.KafkaListenerEndpointContainer#2-0-C-1] - - - [Consumer clientId=consumer-ps-as-xxx-consumer-local-1, groupId=ps-as-xxx-consumer-local] Topic authorization failed for topics [ps-mon.0]
2021-05-07 19:31:52 ERROR o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer:149 [org.springframework.kafka.KafkaListenerEndpointContainer#2-0-C-1] - - - Authorization Exception and no authorizationExceptionRetryInterval set
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [ps-mon.0]
2021-05-07 19:31:52 ERROR o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer:140 [org.springframework.kafka.KafkaListenerEndpointContainer#2-0-C-1] - - - Fatal consumer exception; stopping container
**

Not sure about what is missing here? Can't we have different factories with different credentials to access the topics in the JVM? This is a spring boot application using Spring-kafka. Any pointers would be really appreciated?

[garyrussell]

See the Kafka Documentation.

JAAS configuration using client configuration property

Clients may specify JAAS configuration as a producer or consumer property without creating a physical configuration file. This mode also enables different producers and consumers within the same JVM to use different credentials by specifying different properties for each client. If both static JAAS configuration system property java.security.auth.login.config and client property sasl.jaas.config are specified, the client property will be used.

You can override kafka consumer properties for the factory using ContainerProperties.kafkaConsumerProperties property

/**
* Set the consumer properties that will be merged with the consumer properties
* provided by the consumer factory; properties here will supersede any with the same
* name(s) in the consumer factory.
* {@code group.id} and {@code client.id} are ignored.
* Property keys must be {@link String}s.
* @param kafkaConsumerProperties the properties.
* @see org.apache.kafka.clients.consumer.ConsumerConfig
* @see #setGroupId(String)
* @see #setClientId(String)
*/
public void setKafkaConsumerProperties(Properties kafkaConsumerProperties) {

or at the indidual @KafkaListener

/**
* Kafka consumer properties; they will supersede any properties with the same name
* defined in the consumer factory (if the consumer factory supports property overrides).
* <p>
* <b>Supported Syntax</b>
* <p>The supported syntax for key-value pairs is the same as the
* syntax defined for entries in a Java
* {@linkplain java.util.Properties#load(java.io.Reader) properties file}:
* <ul>
* <li>{@code key=value}</li>
* <li>{@code key:value}</li>
* <li>{@code key value}</li>
* </ul>
* {@code group.id} and {@code client.id} are ignored.
* @return the properties.
* @since 2.2.4
* @see org.apache.kafka.clients.consumer.ConsumerConfig
* @see #groupId()
* @see #clientIdPrefix()
*/
String[] properties() default {};

There is a similar override mechanism in the KafkaTemplate (since 2.5).

[vkpuduran]

Still getting the same exception. Tried to set "setKafkaConsumerProperties" as below.

   @Bean(name = "abcKafkaListenerContainerFactory")
   public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> 
                                                                                      abcKafkaListenerContainerFactory() throws Exception {
	
	ConcurrentKafkaListenerContainerFactory<String, String> factory = kafkaManager.getConcurrentKafkaListenerContainerFactory(getConsumerConfigProperties(abcTopic), numberOfConcurrentListeners, listenerAckMode);
	
              /* setting below properties
               sasl.mechanism SCRAM-SHA-512
               sasl.jaas.config org.apache.kafka.common.security.scram.ScramLoginModule required username="xxx" password="yyy";
               security.protocol SASL_SSL */

	factory.getContainerProperties().setKafkaConsumerProperties(setTopicCredentials(abcTopic));

            //here the username and password are specific to this topic and printed as below
           
    return factory;
}

 @Bean(name = "mnoKafkaListenerContainerFactory")
 public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> 
                                                                                    mnoKafkaListenerContainerFactory() throws Exception {
	
	ConcurrentKafkaListenerContainerFactory<String, String> factory = kafkaManager.getConcurrentKafkaListenerContainerFactory(getConsumerConfigProperties(topic), numberOfConcurrentListeners, listenerAckMode);
	
            /* setting below properties
               sasl.mechanism SCRAM-SHA-512
               sasl.jaas.config org.apache.kafka.common.security.scram.ScramLoginModule required username="aaa" password="bbb";
               security.protocol SASL_SSL
           */
	factory.getContainerProperties().setKafkaConsumerProperties(setTopicCredentials(topic));

         //here the username and password are specific to this topic and printed as below
             
    
    return factory;
}

My consumer code is,

           @KafkaListener(topics = "#{@abcTopic}", groupId = "#{@groupId}", containerFactory = 
                  "abcKafkaListenerContainerFactory")
                   public void receiveEvents(List<ConsumerRecord<String, Object>> records, Acknowledgment ack) throws Exception 
            {
	    
	            List<KafkaRecord> messages = new ArrayList<>();
	            
	            populateMessages(records);
	            
	            ack.acknowledge();
	            
	            processMessages(messages);
	            
	            messages.clear();
            }

        @KafkaListener(topics = "#{@topic}", groupId = "#{@groupId}", containerFactory = "mnoKafkaListenerContainerFactory")
        public void receiveMNOEvents(List<ConsumerRecord<String, Object>> records, Acknowledgment ack) throws Exception {
	        
	        List<KafkaRecord> messages = new ArrayList<>();
	        
	        populateMessages(records);
	        
	        ack.acknowledge();
	        
	        processMessages(messages);
	        
	        messages.clear();
        }

      2021-05-11 15:19:51 ERROR o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer:149 [org.springframework.kafka.KafkaListenerEndpointContainer#3-2-C-1] -  -  -  Authorization Exception and no authorizationExceptionRetryInterval set
      org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [abcTopic]

      2021-05-11 15:20:01 ERROR o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer:149 [org.springframework.kafka.KafkaListenerEndpointContainer#2-0-C-1] -  -  -  Authorization Exception and no authorizationExceptionRetryInterval set
      org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [topic]

[garyrussell]

You can look at the Consumer Config INFO log to see if the properties are applied properly.

[vkpuduran]

    Below details are getting logged, but the "sasl.jaas.config" detail is hidden. Any other way to look at it?

sasl.client.callback.handler.class = null
sasl.jaas.config = [hidden]
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 = SCRAM-SHA-512
security.protocol = SASL_SSL

[garyrussell]

In a debugger, perhaps?

[WachTy]

@vkpuduran Did you manage to solve this issue?

[WachTy]

Our issue was related to a wrong configuration (wrong quotes) in the properties file.