- Properties properties = new Properties();
-
- properties.put(ProducerConfig.CLIENT_ID_CONFIG, deConfiguration.getDistributionStatusTopic().getConsumerId() + "-producer-" + UUID.randomUUID());
- properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
- properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
- properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, deConfiguration.getKafkaBootStrapServers());
- properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
- properties.put(SaslConfigs.SASL_JAAS_CONFIG, getKafkaSaslJaasConfig());
- properties.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
+ KafkaCommonConfig kafkaCommonConfig = new KafkaCommonConfig(deConfiguration);
+ Properties properties = kafkaCommonConfig.getProducerProperties();