@Test public void resetToEarliestTest() throws IOException, InterruptedException, ExecutionException { String topic = "kafka.SimpleTextTopic1"; kafkaCluster.createTopic(topic, 3, 1); TopicMetadata topicMeta = kafkaCluster.waitTopicUntilReady(topic); System.out.println(topicMeta); String group = "SimpleTextTopic1Group"; List<String> expected = new ArrayList<String>(); expected.add("abc"); expected.add("DEF"); expected.add("#$%"); expected.add(" 23"); expected.add("+- "); expected.add(" # "); final List<String> actual = new ArrayList<String>(); Producer producer = Producer.getInstance(); ConsumerHolder consumer = Consumer.getInstance().start(topic, group, new BaseMessageListener<String>() { @Override protected void onMessage(ConsumerMessage<String> msg) { String body = msg.getBody(); actual.add(body); System.out.println("Receive1: " + body); } }); System.out.println("Starting consumer..."); Thread.sleep(CONSUMER_WAIT_BEFORE_READY); for (int i = 0; i < expected.size(); i++) { String proMsg = expected.get(i); MessageHolder holder = producer.message(topic, String.valueOf(i), proMsg); KafkaFuture future = (KafkaFuture) holder.send(); KafkaSendResult result = future.get(); System.out.println(String.format("Sent:%s, Partition:%s, Offset:%s", proMsg, result.getPartition(), result.getOffset())); } int sleepCount = 0; while (actual.size() < expected.size() && sleepCount++ < 50) { Thread.sleep(100); } consumer.close(); Assert.assertEquals(expected.size(), actual.size()); Assert.assertEquals(new HashSet<String>(expected), new HashSet<String>(actual)); KafkaService kafkaService = PlexusComponentLocator.lookup(KafkaService.class); kafkaService.resetConsumerOffset(topic, group, KafkaService.RESET_POSITION.EARLIEST); consumer = Consumer.getInstance().start(topic, group, new BaseMessageListener<String>() { @Override protected void onMessage(ConsumerMessage<String> msg) { String body = msg.getBody(); actual.add(body); System.out.println("Receive1: " + body); } }); System.out.println("Starting consumer..."); Thread.sleep(CONSUMER_WAIT_BEFORE_READY); sleepCount = 0; while (actual.size() < expected.size() * 2 && sleepCount++ < 50) { Thread.sleep(100); } consumer.close(); Assert.assertEquals(expected.size() * 2, actual.size()); Assert.assertEquals(new HashSet<String>(expected), new HashSet<String>(actual)); KafkaMessageSender kafkaSender = (KafkaMessageSender) PlexusComponentLocator.lookup(MessageSender.class, Endpoint.KAFKA); kafkaSender.close(); }
public void resetConsumerOffset(String topic, String consumerGroup, RESET_POSITION position) { Properties prop = getProperties(topic, consumerGroup); KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<String, byte[]>(prop); List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic); List<TopicPartition> topicPartitions = new ArrayList<TopicPartition>(); for (PartitionInfo partitionInfo : partitionInfos) { TopicPartition tp = new TopicPartition(partitionInfo.topic(), partitionInfo.partition()); topicPartitions.add(tp); } consumer.assign(topicPartitions); Set<TopicPartition> assignment = consumer.assignment(); List<PartitionInfo> partitions = consumer.partitionsFor(topic); if (assignment.size() != partitions.size()) { m_logger.warn("ASSIGNMENTS: " + assignment); m_logger.warn("PARTITIONS: " + partitions); m_logger.warn("Could not match, reset failed"); consumer.close(); return; } for (TopicPartition tp : assignment) { long before = consumer.position(tp); if (position == RESET_POSITION.EARLIEST) { consumer.seekToBeginning(tp); } else if (position == RESET_POSITION.LATEST) { consumer.seekToEnd(tp); } long after = consumer.position(tp); m_logger.info("Reset partition: {} From: {} To: {}", tp.partition(), before, after); } consumer.commitSync(); consumer.close(); m_logger.info("Reset offset Done"); }
KafkaService { public void resetConsumerOffset(String topic, String consumerGroup, RESET_POSITION position) { Properties prop = getProperties(topic, consumerGroup); KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<String, byte[]>(prop); List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic); List<TopicPartition> topicPartitions = new ArrayList<TopicPartition>(); for (PartitionInfo partitionInfo : partitionInfos) { TopicPartition tp = new TopicPartition(partitionInfo.topic(), partitionInfo.partition()); topicPartitions.add(tp); } consumer.assign(topicPartitions); Set<TopicPartition> assignment = consumer.assignment(); List<PartitionInfo> partitions = consumer.partitionsFor(topic); if (assignment.size() != partitions.size()) { m_logger.warn("ASSIGNMENTS: " + assignment); m_logger.warn("PARTITIONS: " + partitions); m_logger.warn("Could not match, reset failed"); consumer.close(); return; } for (TopicPartition tp : assignment) { long before = consumer.position(tp); if (position == RESET_POSITION.EARLIEST) { consumer.seekToBeginning(tp); } else if (position == RESET_POSITION.LATEST) { consumer.seekToEnd(tp); } long after = consumer.position(tp); m_logger.info("Reset partition: {} From: {} To: {}", tp.partition(), before, after); } consumer.commitSync(); consumer.close(); m_logger.info("Reset offset Done"); } }
KafkaService { public void resetConsumerOffset(String topic, String consumerGroup, RESET_POSITION position) { Properties prop = getProperties(topic, consumerGroup); KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<String, byte[]>(prop); List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic); List<TopicPartition> topicPartitions = new ArrayList<TopicPartition>(); for (PartitionInfo partitionInfo : partitionInfos) { TopicPartition tp = new TopicPartition(partitionInfo.topic(), partitionInfo.partition()); topicPartitions.add(tp); } consumer.assign(topicPartitions); Set<TopicPartition> assignment = consumer.assignment(); List<PartitionInfo> partitions = consumer.partitionsFor(topic); if (assignment.size() != partitions.size()) { m_logger.warn("ASSIGNMENTS: " + assignment); m_logger.warn("PARTITIONS: " + partitions); m_logger.warn("Could not match, reset failed"); consumer.close(); return; } for (TopicPartition tp : assignment) { long before = consumer.position(tp); if (position == RESET_POSITION.EARLIEST) { consumer.seekToBeginning(tp); } else if (position == RESET_POSITION.LATEST) { consumer.seekToEnd(tp); } long after = consumer.position(tp); m_logger.info("Reset partition: {} From: {} To: {}", tp.partition(), before, after); } consumer.commitSync(); consumer.close(); m_logger.info("Reset offset Done"); }  }
KafkaService { public void resetConsumerOffset(String topic, String consumerGroup, RESET_POSITION position) { Properties prop = getProperties(topic, consumerGroup); KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<String, byte[]>(prop); List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic); List<TopicPartition> topicPartitions = new ArrayList<TopicPartition>(); for (PartitionInfo partitionInfo : partitionInfos) { TopicPartition tp = new TopicPartition(partitionInfo.topic(), partitionInfo.partition()); topicPartitions.add(tp); } consumer.assign(topicPartitions); Set<TopicPartition> assignment = consumer.assignment(); List<PartitionInfo> partitions = consumer.partitionsFor(topic); if (assignment.size() != partitions.size()) { m_logger.warn("ASSIGNMENTS: " + assignment); m_logger.warn("PARTITIONS: " + partitions); m_logger.warn("Could not match, reset failed"); consumer.close(); return; } for (TopicPartition tp : assignment) { long before = consumer.position(tp); if (position == RESET_POSITION.EARLIEST) { consumer.seekToBeginning(tp); } else if (position == RESET_POSITION.LATEST) { consumer.seekToEnd(tp); } long after = consumer.position(tp); m_logger.info("Reset partition: {} From: {} To: {}", tp.partition(), before, after); } consumer.commitSync(); consumer.close(); m_logger.info("Reset offset Done"); }  void resetConsumerOffset(String topic, String consumerGroup, RESET_POSITION position); }
KafkaService { public void resetConsumerOffset(String topic, String consumerGroup, RESET_POSITION position) { Properties prop = getProperties(topic, consumerGroup); KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<String, byte[]>(prop); List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic); List<TopicPartition> topicPartitions = new ArrayList<TopicPartition>(); for (PartitionInfo partitionInfo : partitionInfos) { TopicPartition tp = new TopicPartition(partitionInfo.topic(), partitionInfo.partition()); topicPartitions.add(tp); } consumer.assign(topicPartitions); Set<TopicPartition> assignment = consumer.assignment(); List<PartitionInfo> partitions = consumer.partitionsFor(topic); if (assignment.size() != partitions.size()) { m_logger.warn("ASSIGNMENTS: " + assignment); m_logger.warn("PARTITIONS: " + partitions); m_logger.warn("Could not match, reset failed"); consumer.close(); return; } for (TopicPartition tp : assignment) { long before = consumer.position(tp); if (position == RESET_POSITION.EARLIEST) { consumer.seekToBeginning(tp); } else if (position == RESET_POSITION.LATEST) { consumer.seekToEnd(tp); } long after = consumer.position(tp); m_logger.info("Reset partition: {} From: {} To: {}", tp.partition(), before, after); } consumer.commitSync(); consumer.close(); m_logger.info("Reset offset Done"); }  void resetConsumerOffset(String topic, String consumerGroup, RESET_POSITION position);  }