@Test public void testConsume() throws Exception { Config testConfig = ConfigFactory.parseMap(ImmutableMap.of(ConfigurationKeys.KAFKA_BROKERS, "test")); MockConsumer<String, String> consumer = new MockConsumer<String, String>(OffsetResetStrategy.NONE); consumer.assign(Arrays.asList(new TopicPartition("test_topic", 0))); HashMap<TopicPartition, Long> beginningOffsets = new HashMap<>(); beginningOffsets.put(new TopicPartition("test_topic", 0), 0L); consumer.updateBeginningOffsets(beginningOffsets); ConsumerRecord<String, String> record0 = new ConsumerRecord<>("test_topic", 0, 0L, "key", "value0"); ConsumerRecord<String, String> record1 = new ConsumerRecord<>("test_topic", 0, 1L, "key", "value1"); ConsumerRecord<String, String> record2 = new ConsumerRecord<>("test_topic", 0, 2L, "key", "value2"); consumer.addRecord(record0); consumer.addRecord(record1); consumer.addRecord(record2); try (Kafka09ConsumerClient<String, String> kafka09Client = new Kafka09ConsumerClient<>(testConfig, consumer);) { Set<KafkaConsumerRecord> consumedRecords = Sets.newHashSet(kafka09Client.consume(new KafkaPartition.Builder().withId(0).withTopicName("test_topic") .build(), 0l, 100l)); Set<Kafka09ConsumerRecord<String, String>> expected = ImmutableSet.<Kafka09ConsumerRecord<String, String>> of(new Kafka09ConsumerRecord<>(record0), new Kafka09ConsumerRecord<>(record1), new Kafka09ConsumerRecord<>(record2)); Assert.assertEquals(consumedRecords, expected); } }
@Override public Iterator<KafkaConsumerRecord> consume(KafkaPartition partition, long nextOffset, long maxOffset) { if (nextOffset > maxOffset) { return null; } this.consumer.assign(Lists.newArrayList(new TopicPartition(partition.getTopicName(), partition.getId()))); this.consumer.seek(new TopicPartition(partition.getTopicName(), partition.getId()), nextOffset); return consume(); }
Kafka09ConsumerClient extends AbstractBaseKafkaConsumerClient { @Override public Iterator<KafkaConsumerRecord> consume(KafkaPartition partition, long nextOffset, long maxOffset) { if (nextOffset > maxOffset) { return null; } this.consumer.assign(Lists.newArrayList(new TopicPartition(partition.getTopicName(), partition.getId()))); this.consumer.seek(new TopicPartition(partition.getTopicName(), partition.getId()), nextOffset); return consume(); } }
Kafka09ConsumerClient extends AbstractBaseKafkaConsumerClient { @Override public Iterator<KafkaConsumerRecord> consume(KafkaPartition partition, long nextOffset, long maxOffset) { if (nextOffset > maxOffset) { return null; } this.consumer.assign(Lists.newArrayList(new TopicPartition(partition.getTopicName(), partition.getId()))); this.consumer.seek(new TopicPartition(partition.getTopicName(), partition.getId()), nextOffset); return consume(); } private  Kafka09ConsumerClient(Config config);  Kafka09ConsumerClient(Config config, Consumer<K, V> consumer); }
Kafka09ConsumerClient extends AbstractBaseKafkaConsumerClient { @Override public Iterator<KafkaConsumerRecord> consume(KafkaPartition partition, long nextOffset, long maxOffset) { if (nextOffset > maxOffset) { return null; } this.consumer.assign(Lists.newArrayList(new TopicPartition(partition.getTopicName(), partition.getId()))); this.consumer.seek(new TopicPartition(partition.getTopicName(), partition.getId()), nextOffset); return consume(); } private  Kafka09ConsumerClient(Config config);  Kafka09ConsumerClient(Config config, Consumer<K, V> consumer); @Override List<KafkaTopic> getTopics(); @Override long getEarliestOffset(KafkaPartition partition); @Override long getLatestOffset(KafkaPartition partition); @Override Iterator<KafkaConsumerRecord> consume(KafkaPartition partition, long nextOffset, long maxOffset); @Override Iterator<KafkaConsumerRecord> consume(); @Override void subscribe(String topic); @Override void subscribe(String topic, GobblinConsumerRebalanceListener listener); @Override Map<String, Metric> getMetrics(); @Override void commitOffsetsAsync(Map<KafkaPartition, Long> partitionOffsets); @Override void commitOffsetsSync(Map<KafkaPartition, Long> partitionOffsets); @Override long committed(KafkaPartition partition); @Override void close(); }
Kafka09ConsumerClient extends AbstractBaseKafkaConsumerClient { @Override public Iterator<KafkaConsumerRecord> consume(KafkaPartition partition, long nextOffset, long maxOffset) { if (nextOffset > maxOffset) { return null; } this.consumer.assign(Lists.newArrayList(new TopicPartition(partition.getTopicName(), partition.getId()))); this.consumer.seek(new TopicPartition(partition.getTopicName(), partition.getId()), nextOffset); return consume(); } private  Kafka09ConsumerClient(Config config);  Kafka09ConsumerClient(Config config, Consumer<K, V> consumer); @Override List<KafkaTopic> getTopics(); @Override long getEarliestOffset(KafkaPartition partition); @Override long getLatestOffset(KafkaPartition partition); @Override Iterator<KafkaConsumerRecord> consume(KafkaPartition partition, long nextOffset, long maxOffset); @Override Iterator<KafkaConsumerRecord> consume(); @Override void subscribe(String topic); @Override void subscribe(String topic, GobblinConsumerRebalanceListener listener); @Override Map<String, Metric> getMetrics(); @Override void commitOffsetsAsync(Map<KafkaPartition, Long> partitionOffsets); @Override void commitOffsetsSync(Map<KafkaPartition, Long> partitionOffsets); @Override long committed(KafkaPartition partition); @Override void close(); static final String KAFKA_09_DEFAULT_KEY_DESERIALIZER; static final String GOBBLIN_CONFIG_KEY_DESERIALIZER_CLASS_KEY; static final String GOBBLIN_CONFIG_VALUE_DESERIALIZER_CLASS_KEY; }