0
I created a class called ConsumerConfig and Service class that contain fonction that gets records from a topic.
locally it works just fine but when i add the brokers it stopped working and i get timout exception, and it takes a loong time can anybody help me .
here is the code of the ConsumerConfig class and the fonctions that gets records from specific topic :
@Component
public class ConsumerConfig {
private static Integer numberOfConsumer = 0;
private KafkaConsumer consumer;
private Map<String, Object> buildDefaultConfig() {
final Map<String, Object> defaultClientConfig = new HashMap<>();
defaultClientConfig.put("bootstrap.servers", (String) getbrokersConfigurationFromFile().get("spring.kafka.bootstrap-servers"));
defaultClientConfig.put("client.id", "test-consumer-id" + (++numberOfConsumer));
defaultClientConfig.put("request.timeout.ms",60000);
return defaultClientConfig;
}
@Bean
@RequestScope
public <K, V> KafkaConsumer<K, V> getKafkaConsumer() {
// Build config
final Map<String, Object> kafkaConsumerConfig = buildDefaultConfig();
kafkaConsumerConfig.put("key.deserializer", StringDeserializer.class);
kafkaConsumerConfig.put("value.deserializer", StringDeserializer.class);
kafkaConsumerConfig.put("auto.offset.reset", "earliest");
kafkaConsumerConfig.put("max.poll.records",5);
kafkaConsumerConfig.put("default.api.timeout.ms", 6000000);
kafkaConsumerConfig.put("max.block.ms ",60000000);
kafkaConsumerConfig.put("auto.offset.reset", "earliest");
kafkaConsumerConfig.put("enable.partition.eof", "false");
kafkaConsumerConfig.put("enable.auto.commit", "true");
kafkaConsumerConfig.put("auto.commit.interval.ms", "1000");
kafkaConsumerConfig.put("session.timeout.ms", "30000");
//fetch.max.byte The maximum amount of data the server should return for a fetch request.
// Create and return Consumer.
return consumer=new KafkaConsumer<K, V> (kafkaConsumerConfig);
}
}
/**
* reacd from specific offset timestamp
*/
public <K, V> List<Record> consumeFromTime(String topicName, Long timestampMs) {
// Get the list of partitions
List<PartitionInfo> partitionInfos = consumer.partitionsFor(topicName);
// Transform PartitionInfo into TopicPartition
List<TopicPartition> topicPartitionList = partitionInfos
.stream()
.map(info -> new TopicPartition(topicName, info.partition()))
.collect(Collectors.toList());
// Assign the consumer to these partitions
consumer.assign(topicPartitionList);
// Look for offsets based on timestamp
Map<TopicPartition, Long> partitionTimestampMap = topicPartitionList.stream()
.collect(Collectors.toMap(tp -> tp, tp -> System.currentTimeMillis() - timestampMs * 1000));
Map<TopicPartition, OffsetAndTimestamp> partitionOffsetMap = consumer.offsetsForTimes(partitionTimestampMap);
// Force the consumer to seek for those offsets
List<ConsumerRecord<K, V>> allRecords = new ArrayList<>();
partitionOffsetMap.forEach((tp, offsetAndTimestamp) -> {
if (!Objects.isNull(offsetAndTimestamp)) {
consumer.seek(tp, offsetAndTimestamp.offset());
ConsumerRecords<K, V> records;
records = consumer.poll(Duration.ofMillis(100L));
records.forEach(allRecords::add);
}
}
);
return allRecords.stream().map(v -> Record.builder().values(v.value()).offset(v.offset()).partition(v.partition()).build()).collect(Collectors.toList());
}