package mib.microservice.commons.kafka;

import mib.microservice.commons.events.IEventConsumer;
import mib.microservice.commons.events.base.EventBase;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.message.MessageAndMetadata;


public class Consumer<K, V extends EventBase<?>> implements Runnable {
	private KafkaStream<K, V> stream;
	private IEventConsumer<K, V> command;
	
	public Consumer(KafkaStream<K, V> stream, IEventConsumer<K, V> command) {
		this.stream = stream;
		this.command = command;
	}
	
	@Override
	public void run() {
		ConsumerIterator<K, V> it = this.stream.iterator();
		while (it.hasNext()) {
			MessageAndMetadata<K, V> msg = it.next();
			
			System.out.println("p" + msg.partition() + " @" + msg.offset());
			
			this.command.execute(msg.key(), msg.message());
		}
	}
}
@@ -34,16 +34,92 @@
 kafka;%0A%0A
+import kafka.consumer.KafkaStream;%0Aimport kafka.message.MessageAndMetadata;%0A
 import m
@@ -224,124 +224,8 @@
 se;%0A
-import kafka.consumer.ConsumerIterator;%0Aimport kafka.consumer.KafkaStream;%0Aimport kafka.message.MessageAndMetadata;%0A
 %0A%0Apu
@@ -537,88 +537,13 @@
 %7B%0A%09%09
-ConsumerIterator%3CK, V%3E it = this.stream.iterator();%0A%09%09while (it.hasNext()) %7B%0A%09%09%09
+for (
 Mess
@@ -571,24 +571,24 @@
 msg 
-= it.next();%0A%09%09%09
+: this.stream) %7B
 %0A%09%09%09

Use for-loop instead of while+iterator