Optimize consumer

This commit is contained in:
Andrea Cavalli 2022-09-10 13:36:38 +02:00
parent 0055c54826
commit aa12013479

View File

@ -49,6 +49,8 @@ public abstract class KafkaConsumer<K> {
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, getChannelName().getDeserializerClass());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, toIntExact(Duration.ofMinutes(5).toMillis()));
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1048576");
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "100");
ReceiverOptions<Integer, K> receiverOptions = ReceiverOptions
.<Integer, K>create(props)
.commitInterval(Duration.ofSeconds(10))