KafkaOffsetManager implements AutoCloseable { public void setOffsets(Map<TopicAndPartition, Long> offsets) throws IllegalArgumentException { if (MapUtils.isEmpty(offsets)) { throw new IllegalArgumentException("Offsets map can't be null or empty"); } if (!KafkaOffsetManagerUtils.isValidOffsetMap(offsets)) { throw new IllegalArgumentException("Invalid offsets Map<TopicAndPartition, Long>"); } dao.setOffsets(groupid, offsets); } private  KafkaOffsetManager(); void close(); Map<TopicAndPartition, Long> getOffsets(); void setOffsets(Map<TopicAndPartition, Long> offsets); Map<TopicAndPartition, Long> getEarliestOffsets(); Map<TopicAndPartition, Long> getEarliestOffsets(String topic); Map<TopicAndPartition, Long> getLatestOffsets(); Map<TopicAndPartition, Long> getLatestOffsets(String topic);  }
@Test public void testSetOffSetsOK() throws Exception { for (String topic : topicList) { Map<TopicAndPartition, Long> tap = new HashMap<>(); tap.put(new TopicAndPartition(topic, 1), 1L); tap.put(new TopicAndPartition(topic, 2), 2L); validKOM.setOffsets(tap); } }
@Test(expected = IllegalArgumentException.class) public void testSetOffsetsWithInvalidParam() throws Exception { Map<TopicAndPartition, Long> tap = new HashMap<>(); validKOM.setOffsets(tap); }
@Test(expected = IllegalArgumentException.class) public void testSetWithInvalidParams() throws Exception { validKOM.setOffsets(null); }
@Test(expected = IllegalArgumentException.class) public void testSetOffSetsWrongMap() throws Exception { for (String topic : topicList) { Map<TopicAndPartition, Long> tap = new HashMap<>(); tap.put(new TopicAndPartition(topic, 1), 1L); tap.put(new TopicAndPartition(topic + "1", 2), 2L); validKOM.setOffsets(tap); } }