[KAFKA-3972] kafka java consumer poll returns 0 records after seekToBeginning

• Type: Task
• Status: Resolved
• Priority: Major
• Resolution: Invalid
• Affects Version/s: 0.10.0.0
• Component/s: consumer
• Labels: kafka polling
• Environment: docker image elasticsearch:latest, kafka scala version 2.11, kafka version 0.10.0.0
• Assignee: Ewen Cheslack-Postava
• Reporter: don caldwell
• Created: 18/Jul/16 20:33
• Updated: 08/Aug/16 06:43
• Resolved: 08/Aug/16 06:43

Description

kafkacat successfully returns rows for the topic, but the following java source reliably fails to produce rows. I have the suspicion that I am missing some simple thing in my setup, but I have been unable to find a way out. I am using the current docker and using docker network commands to connect the processes in my cluster. The properties are:
bootstrap.servers: kafka01:9092,kafka02:9092,kafka03:9092
group.id: dhcp1
topic: dhcp
enable.auto.commit: false
auto.commit.interval.ms: 1000
session.timeout.ms 30000
key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
value.deserializer: org.apache.kafka.common.serialization.StringDeserializer

the kafka consumer follows. One thing that I find curious is that, although I seem to successfully make the call to seekToBeginning(), when I print offsets on failure, I get large offsets for all partitions although I had expected them to be 0 or at least some small number.
Here is the code:

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.protocol.types.SchemaException;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.Integer;
import java.lang.System;
import java.lang.Thread;
import java.lang.InterruptedException;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class KConsumer {
private Properties prop;
private String topic;
private Integer polln;
private KafkaConsumer<String, String> consumer;
private String[] pna =

{ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ConsumerConfig.GROUP_ID_CONFIG, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG}

;

public KConsumer(String pf) throws FileNotFoundException,
IOException

{ this.setProperties(pf); this.newClient(); }

public void setProperties(String p) throws FileNotFoundException,
IOException

{ this.prop = new Properties(); this.prop.load(new FileInputStream(p)); this.topic = this.prop.getProperty("topic"); this.polln = new Integer(this.prop.getProperty("polln")); }

public void setTopic(String t)

{ this.topic = t; }

public String getTopic()

{ return this.topic; }

public void newClient() {
System.err.println("creating consumer");
Properties kp = new Properties();
for(String p : pna) {
String v = this.prop.getProperty(p);
if(v != null)

{ kp.put(p, v); }

}
//this.consumer = new KafkaConsumer<>(this.prop);
this.consumer = new KafkaConsumer<>(kp);
//this.consumer.subscribe(Collections.singletonList(this.topic));
System.err.println("subscribing to " + this.topic);
this.consumer.subscribe(Arrays.asList(this.topic));
//this.seekToBeginning();
}

public void close()

{ this.consumer.close(); this.consumer = null; }

public void seekToBeginning() {
if(this.topic == null)

{ System.err.println("KConsumer: topic not set"); System.exit(1); }

System.err.println("setting partition offset to beginning");
java.util.Set<TopicPartition> tps = this.consumer.assignment();
this.consumer.seekToBeginning(tps);
}

public ConsumerRecords<String,String> nextBatch()
throws KafkaException {
while(true) {
try

{ System.err.printf("polling..."); ConsumerRecords<String,String> records = this.consumer.poll(this.polln); System.err.println("returned"); System.err.printf("record count %d\n", records.count()); return records; }

catch(SchemaException e)

{ System.err.println("nextBatch: " + e); }

catch(KafkaException e)

{ System.err.println("nextBatch: " + e); throw e; }

catch(Exception e)

{ System.err.println("nextBatch: " + e); this.consumer.close(); System.exit(1); }

try

{ System.err.println("sleeping"); Thread.sleep(2000); }

catch(InterruptedException e)

{ System.err.println(e); System.exit(0); }

}
}

public void printBatch(ConsumerRecords<String,String> records) {
System.err.println("printing...");
Iterable<ConsumerRecord<String,String>> ri =
records.records(this.topic);
for (ConsumerRecord<String, String> record : ri)

{ System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); }

}

public void doProcess() {
Integer n = 0;
Integer f = 0;
long total = 0;
try {
while(true) {
ConsumerRecords<String,String> r = this.nextBatch();
long count = r.count();
if(r.count() > 0)

{ total += count; this.printBatch(r); n = n + 1; }

else

{ f = f + 1; }

if(f > 10)

{ System.err.printf("total %d\n", total); this.printMisc(); break; }

}
} finally

{ this.consumer.close(); }

}

public void printPosition(int pid) {
try

{ TopicPartition tp = new TopicPartition(this.topic, pid); long pos = this.consumer.position(tp); System.err.printf(" offset: %d\n", pos); }

catch(IllegalArgumentException e)

{ System.err.printf("printPosition: %d %s\n", pid, e); }

}
public void printMisc() {
Map<String,List<PartitionInfo>> topicMap;
List<PartitionInfo> partitionList;

System.err.println("in printMisc");
try {
topicMap = this.consumer.listTopics();
for(String key: topicMap.keySet()) {
if(key.compareTo(this.topic) != 0) continue;
System.err.printf("topic: %s\n", key);
List<PartitionInfo> pl = topicMap.get(key);
for(PartitionInfo pinf: pl) {
System.err.printf("partition %d\n", pinf.partition());
System.err.printf(" leader %s\n",
pinf.leader().host());
this.printPosition(pinf.partition());
System.err.printf(" replicas:\n");
for(Node r: pinf.replicas())

{ System.err.printf(" %s %s\n", r.id(), r.host()); }
System.err.printf(" inSyncReplicas:\n");
for(Node r: pinf.inSyncReplicas()) { System.err.printf(" %s %sn", r.id(), r.host()); }

}
}
} catch (TimeoutException e)

{ System.err.printf("printMisc: %s\n", e); //System.exit(1); }

}

public static void main(String[] args) throws FileNotFoundException,
IOException, InterruptedException {
if(args.length == 1)

{ Thread.sleep(2000); // docker network connect KConsumer kc = new KConsumer(args[0]); //kc.printMisc(); kc.doProcess(); }

else

{ System.err.println("Usage KConsumer propfile"); System.exit(1); }

}
}