Kafka Consumer For Partitions

kafka-consumer-for-partion.java

List<String> consumer(String topic) {
        List<String> events = new ArrayList<>();
        List<PartitionInfo> partitionInfos = null;
        List<TopicPartition> partitions = new ArrayList<>();
        partitionInfos = consumer.partitionsFor(topic);

        if (partitionInfos != null) {
            for (PartitionInfo partition : partitionInfos) 
                partitions.add(new TopicPartition(partition.topic(),
                        partition.partition()));
            consumer.assign(partitions);
            

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(1000);

                for (ConsumerRecord<String, String> record : records) {
                    events.add(record.value());
                }
                consumer.commitSync();

                if (!records.isEmpty()) {
                    break;
                }
            }
        }

        return events;
    }