@Test public void testNextKeyValue() throws Throwable { KafkaInputRecordReader kafkaInputRecordReader = new KafkaInputRecordReader(); assertFalse(kafkaInputRecordReader.nextKeyValue()); assertFalse(kafkaInputRecordReader.nextKeyValue()); }
@Override public boolean nextKeyValue() throws IOException, InterruptedException { if (key == null) { key = new LongWritable(); } if (value == null) { value = new BytesWritable(); } if (watermark >= latestOffset) { log.info("Reach the end offset, stop reading."); return false; } if (messages == null) { log.info("{} fetching offset {} ", topic + ":" + split.getBrokers() + ":" + partition, watermark); TopicPartition topicPartition = new TopicPartition(topic, partition); consumer.seek(topicPartition, watermark); messages = consumer.poll(timeOut); iterator = messages.iterator(); if (!iterator.hasNext()) { log.info("No more messages, stop"); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset " + "%d, but end at %d", latestOffset, watermark)); } } if (iterator.hasNext()) { ConsumerRecord<String, String> message = iterator.next(); key.set(message.offset()); byte[] valuebytes = Bytes.toBytes(message.value()); value.set(valuebytes, 0, valuebytes.length); watermark = message.offset() + 1; numProcessedMessages++; if (!iterator.hasNext()) { messages = null; iterator = null; } return true; } log.error("Unexpected iterator end."); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset %d, but end at %d", latestOffset, watermark)); }
KafkaInputRecordReader extends RecordReader<LongWritable, BytesWritable> { @Override public boolean nextKeyValue() throws IOException, InterruptedException { if (key == null) { key = new LongWritable(); } if (value == null) { value = new BytesWritable(); } if (watermark >= latestOffset) { log.info("Reach the end offset, stop reading."); return false; } if (messages == null) { log.info("{} fetching offset {} ", topic + ":" + split.getBrokers() + ":" + partition, watermark); TopicPartition topicPartition = new TopicPartition(topic, partition); consumer.seek(topicPartition, watermark); messages = consumer.poll(timeOut); iterator = messages.iterator(); if (!iterator.hasNext()) { log.info("No more messages, stop"); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset " + "%d, but end at %d", latestOffset, watermark)); } } if (iterator.hasNext()) { ConsumerRecord<String, String> message = iterator.next(); key.set(message.offset()); byte[] valuebytes = Bytes.toBytes(message.value()); value.set(valuebytes, 0, valuebytes.length); watermark = message.offset() + 1; numProcessedMessages++; if (!iterator.hasNext()) { messages = null; iterator = null; } return true; } log.error("Unexpected iterator end."); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset %d, but end at %d", latestOffset, watermark)); } }
KafkaInputRecordReader extends RecordReader<LongWritable, BytesWritable> { @Override public boolean nextKeyValue() throws IOException, InterruptedException { if (key == null) { key = new LongWritable(); } if (value == null) { value = new BytesWritable(); } if (watermark >= latestOffset) { log.info("Reach the end offset, stop reading."); return false; } if (messages == null) { log.info("{} fetching offset {} ", topic + ":" + split.getBrokers() + ":" + partition, watermark); TopicPartition topicPartition = new TopicPartition(topic, partition); consumer.seek(topicPartition, watermark); messages = consumer.poll(timeOut); iterator = messages.iterator(); if (!iterator.hasNext()) { log.info("No more messages, stop"); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset " + "%d, but end at %d", latestOffset, watermark)); } } if (iterator.hasNext()) { ConsumerRecord<String, String> message = iterator.next(); key.set(message.offset()); byte[] valuebytes = Bytes.toBytes(message.value()); value.set(valuebytes, 0, valuebytes.length); watermark = message.offset() + 1; numProcessedMessages++; if (!iterator.hasNext()) { messages = null; iterator = null; } return true; } log.error("Unexpected iterator end."); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset %d, but end at %d", latestOffset, watermark)); }  }
KafkaInputRecordReader extends RecordReader<LongWritable, BytesWritable> { @Override public boolean nextKeyValue() throws IOException, InterruptedException { if (key == null) { key = new LongWritable(); } if (value == null) { value = new BytesWritable(); } if (watermark >= latestOffset) { log.info("Reach the end offset, stop reading."); return false; } if (messages == null) { log.info("{} fetching offset {} ", topic + ":" + split.getBrokers() + ":" + partition, watermark); TopicPartition topicPartition = new TopicPartition(topic, partition); consumer.seek(topicPartition, watermark); messages = consumer.poll(timeOut); iterator = messages.iterator(); if (!iterator.hasNext()) { log.info("No more messages, stop"); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset " + "%d, but end at %d", latestOffset, watermark)); } } if (iterator.hasNext()) { ConsumerRecord<String, String> message = iterator.next(); key.set(message.offset()); byte[] valuebytes = Bytes.toBytes(message.value()); value.set(valuebytes, 0, valuebytes.length); watermark = message.offset() + 1; numProcessedMessages++; if (!iterator.hasNext()) { messages = null; iterator = null; } return true; } log.error("Unexpected iterator end."); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset %d, but end at %d", latestOffset, watermark)); }  @Override void initialize(InputSplit split, TaskAttemptContext context); void initialize(InputSplit split, Configuration conf); @Override boolean nextKeyValue(); @Override LongWritable getCurrentKey(); @Override BytesWritable getCurrentValue(); @Override float getProgress(); @Override void close(); }
KafkaInputRecordReader extends RecordReader<LongWritable, BytesWritable> { @Override public boolean nextKeyValue() throws IOException, InterruptedException { if (key == null) { key = new LongWritable(); } if (value == null) { value = new BytesWritable(); } if (watermark >= latestOffset) { log.info("Reach the end offset, stop reading."); return false; } if (messages == null) { log.info("{} fetching offset {} ", topic + ":" + split.getBrokers() + ":" + partition, watermark); TopicPartition topicPartition = new TopicPartition(topic, partition); consumer.seek(topicPartition, watermark); messages = consumer.poll(timeOut); iterator = messages.iterator(); if (!iterator.hasNext()) { log.info("No more messages, stop"); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset " + "%d, but end at %d", latestOffset, watermark)); } } if (iterator.hasNext()) { ConsumerRecord<String, String> message = iterator.next(); key.set(message.offset()); byte[] valuebytes = Bytes.toBytes(message.value()); value.set(valuebytes, 0, valuebytes.length); watermark = message.offset() + 1; numProcessedMessages++; if (!iterator.hasNext()) { messages = null; iterator = null; } return true; } log.error("Unexpected iterator end."); throw new IOException(String.format(Locale.ROOT, "Unexpected ending of stream, expected ending offset %d, but end at %d", latestOffset, watermark)); }  @Override void initialize(InputSplit split, TaskAttemptContext context); void initialize(InputSplit split, Configuration conf); @Override boolean nextKeyValue(); @Override LongWritable getCurrentKey(); @Override BytesWritable getCurrentValue(); @Override float getProgress(); @Override void close(); static final long DEFAULT_KAFKA_CONSUMER_POLL_TIMEOUT; }