treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

kafka settings constants

Browse source
This commit is contained in:
ialbert 2022-08-23 17:42:17 +03:00
parent 57072f3cff
commit 6bc177d506
1 changed files with 7 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/KafkaProducerFactory.java
View file

@ -13,22 +13,22 @@ public class KafkaProducerFactory {
Properties kafkaProps = new Properties(); Properties kafkaProps = new Properties();
//Assign localhost id //Assign localhost id
kafkaProps.put("bootstrap.servers", kafkaSettings.getBootstrapServers()); kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaSettings.getBootstrapServers());
//Set acknowledgements for producer requests. //Set acknowledgements for producer requests.
kafkaProps.put("acks", kafkaSettings.getAcks()); kafkaProps.put(ProducerConfig.ACKS_CONFIG, kafkaSettings.getAcks());
//If the request fails, the producer can automatically retry, //If the request fails, the producer can automatically retry,
kafkaProps.put("retries", kafkaSettings.getRetries()); kafkaProps.put(ProducerConfig.RETRIES_CONFIG, kafkaSettings.getRetries());
//Specify buffer size in config //Specify buffer size in config
kafkaProps.put("batch.size", kafkaSettings.getBatchSize()); kafkaProps.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaSettings.getBatchSize());
//Reduce the no of requests less than 0 //Reduce the no of requests less than 0
kafkaProps.put("linger.ms", kafkaSettings.getLingerMs()); kafkaProps.put(ProducerConfig.LINGER_MS_CONFIG, kafkaSettings.getLingerMs());
//The buffer.memory controls the total amount of memory available to the producer for buffering. //The buffer.memory controls the total amount of memory available to the producer for buffering.
kafkaProps.put("buffer.memory", kafkaSettings.getBufferMemory()); kafkaProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaSettings.getBufferMemory());
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName()); kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());
// kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); //ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());