fun <K, V> readKeyValues( topic: String?, consumerConfig: Properties?, maxMessages: Int ): List<KeyValue<K, V>>? { val consumer: KafkaConsumer<K, V> = KafkaConsumer(consumerConfig) consumer.subscribe(Collections.singletonList(topic)) val pollIntervalMs = 100 val maxTotalPollTimeMs = 2000 var totalPollTimeMs = 0 val consumedValues: MutableList<KeyValue<K, V>> = ArrayList() while (totalPollTimeMs < maxTotalPollTimeMs && continueConsuming( consumedValues.size, maxMessages ) ) { totalPollTimeMs += pollIntervalMs val records: ConsumerRecords<K, V> = consumer.poll(pollIntervalMs) for (record in records) { consumedValues.add(KeyValue(record.key(), record.value())) } } consumer.close() return consumedValues }
Returns up to ` maxMessages ` by reading via the provided consumer ( the topic ( s ) to read from are already configured in the consumer ) .