i have a spring boot application which is trying to connect to two different bootstrap servers,one is a secured server and the other one is a non secured.My KafkaConsumerConfig has two different set of consumerFactory one which is avro based and other is non avro based.The problem here is my avro consumer factory wants to connect to non secure kafka bootstrap server and the other consumer factory wants to connect to secure kafka bootstrap server,i am not able to achieve it.
@Configuration
@ConditionalOnProperty(name = "messaging.consumer.kafka.enabled")
@RequiredArgsConstructor
public class KafkaConsumerConfig implements Loggable {
private static final String KAFKA_CLIENT_ID = "service";
@Autowired
private final KafkaCommonProperties kafkaCommonProperties;
@Autowired
private final KafkaConsumerProperties kafkaConsumerProperties;
public Map<String, Object> plainKafkaConfiguration() {
Map<String, Object> config = new HashMap<>();
// basic config
config.put(BOOTSTRAP_SERVERS_CONFIG, kafkaCommonProperties.getBootstrapServers());
//above line is for secure bootstrap server
config.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(CLIENT_ID_CONFIG, KAFKA_CLIENT_ID);
config.put(GROUP_ID_CONFIG, KAFKA_CLIENT_ID);
config.put(ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerProperties.isAutoCommitEnabled());
config.put(AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaConsumerProperties.getAutoCommitIntervalMs());
config.put(AUTO_OFFSET_RESET_CONFIG, kafkaConsumerProperties.getAutoOffsetReset());
config.put(HEARTBEAT_INTERVAL_MS_CONFIG, kafkaConsumerProperties.getHeartBeatIntervalMs());
return config;
}
public Map<String, Object> avroConfigsConsumerFactory() {
Map<String, Object> config = new HashMap<>();
config.put(BOOTSTRAP_SERVERS_CONFIG,
kafkaCommonProperties.getBootstrapServersNonSecure());
//above line is for non secure kafka bootstrap server
config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
kafkaConsumerProperties.getSchemaRegistryUrl());
config.put(KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
config.put(VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
config.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS,
kafkaConsumerProperties.getPropertiesKeyDeserializer());
config.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS,
kafkaConsumerProperties.getPropertiesValueDeserializer());
config.put(GROUP_ID_CONFIG, KAFKA_CLIENT_ID);
config.put(AUTO_OFFSET_RESET_CONFIG, kafkaConsumerProperties.getAutoOffsetReset());
config.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG,
kafkaConsumerProperties.isPropertiesSpecificAvroReader());
config.put(ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerProperties.isAutoCommitEnabled());
config.put(AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaConsumerProperties.getAutoCommitIntervalMs());
return config;
}
public Map<String, Object> secureKafkaConfiguration() {
Map<String, Object> config = plainKafkaConfiguration();
// secure MSK config
config.put(SECURITY_PROTOCOL_CONFIG, SSL.name);
config.put(SSL_KEYSTORE_LOCATION_CONFIG, kafkaCommonProperties.getKeystorePath());
config.put(SSL_KEYSTORE_PASSWORD_CONFIG, kafkaCommonProperties.getKeystorePassword());
return config;
}
@Bean
@Profile("!test")
public ConsumerFactory<String, String> consumerFactory() {
getLogger().debug("Using secureKafkaConfiguration ...");
return new DefaultKafkaConsumerFactory<>(secureKafkaConfiguration());
}
@Bean
@Profile("test")
public ConsumerFactory<String, String> testConsumerFactory() {
getLogger().debug("Using plainKafkaConfiguration ...");
return new DefaultKafkaConsumerFactory<>(plainKafkaConfiguration());
}
@SuppressWarnings({"rawtypes", "unchecked"})
@Bean("kafkaListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>>
kafkaListenerContainerFactory(ConsumerFactory consumerFactory) {
ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory =
new ConcurrentKafkaListenerContainerFactory<>();
containerFactory.setConsumerFactory(consumerFactory);
containerFactory.setConcurrency(kafkaConsumerProperties.getConcurrentListeners());
getLogger().debug("containerFactory: {}", containerFactory);
return containerFactory;
}
public ConsumerFactory<Object, Object> consumerAvroFactory() {
getLogger().debug("Using Avro kafka configuration ...");
return new DefaultKafkaConsumerFactory<>(avroConfigsConsumerFactory());
}
@Bean("kafkaAvroListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>>
kafkaAvroListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerAvroFactory());
factory.setConcurrency(kafkaConsumerProperties.getConcurrentListeners());
factory.getContainerProperties().setPollTimeout(kafkaConsumerProperties.getPropertiesPollTimeout());
getLogger().debug("containerFactory: {}", factory);
return factory;
}
}
<-------------------------------------------->
@Getter
@ConfigurationProperties(prefix = "messaging.common.kafka")
public class KafkaCommonProperties implements Loggable {
private String bootstrapServers;
private String bootstrapServersNonSecure;
@Value("${messaging.common.kafka.ssl.keystore.path}")
private String keystorePath;
@Value("${messaging.common.kafka.ssl.keystore.password}")
private String keystorePassword;
public void setBootstrapServers(String bootstrapServers) {
this.bootstrapServers = bootstrapServers;
getLogger().debug("bootstrapServers: {}", bootstrapServers);
}
public void setKeystorePath(String keystorePath) {
this.keystorePath = keystorePath;
getLogger().debug("keystorePath: {}", keystorePath);
}
public void setBootstrapServersNonSecure(String bootstrapServersNonSecure) {
getLogger().debug("bootstrapServers non secure: {}", bootstrapServersNonSecure);
this.bootstrapServersNonSecure = bootstrapServersNonSecure;
}
public void setKeystorePassword(String keystorePassword) {
this.keystorePassword = keystorePassword;
}
}