Spring Boot App not connection to Kafka with SSL

Viewed 162

I'm trying to configure kafka in my application and I'm getting the following error. Could you please help me? Ref: https://docs.confluent.io/3.3.1/kafka/ssl.html

org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed Caused by: javax.net.ssl.SSLHandshakeException: General SSLEngine problem

spring:
  application:
    name: application
  autoconfigure:
    exclude: org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
    ssl:
      trust-store-location: src/main/resources/kafka.client.truststore.jks
      trust-store-password: password
      trust-store-type: JKS
      key-store-location: src/main/resources/kafka.client.keystore.jks
      key-store-password: password
      key-store-type: JKS
      protocol: SSL

eureka:
  client:
    registerWithEureka: false
    fetchRegistry: false```

My consumer config

    package config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.security.plain.PlainLoginModule;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.consumer.bootstrap-servers}")
    private String bootstrapServers;

    private static String KAFKA_USER = "user";
    private static String KAFKA_PASSWORD = "password";

    @Bean
    public Map<String, Object> consumerConfigs() {

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "MATHEUS");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        props.put("security.protocol", "SASL_SSL");
        props.put("sasl.mechanism", "PLAIN");
        props.put("sasl.jaas.config", PlainLoginModule.class.getName() + " required username=\"" + KAFKA_USER + "\" password=\"" + KAFKA_PASSWORD + "\";");
        return props;
    }

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
                new JsonDeserializer<>());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

}
0 Answers
Related