import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;
public class SimpleConsumer
{
private long consumed = 0;
- private KafkaConsumer<String, String> consumer;
+ private Consumer<String, String> consumer;
private Lock lock = new ReentrantLock();
private Condition stopped = lock.newCondition();
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
- KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+ Consumer<String, String> consumer = new KafkaConsumer<>(props);
// end::create[]
this.consumer = consumer;
}