KafkaOffsetManagerUtils { static boolean isValidOffsetMap(Map<TopicAndPartition, Long> offsets) { assert(offsets!=null); Set<String> topicSet = new HashSet<>(); for (Map.Entry<TopicAndPartition, Long> entry : offsets.entrySet()) { TopicAndPartition tap = entry.getKey(); topicSet.add(tap.topic()); if (tap.partition()<0) { Log.error("partition must be >=0"); return false; } Long offset = entry.getValue(); if (offset<0L) { Log.error("invalid offset map. offset needs to be >=0"); return false; } } if (topicSet.size()!=1) { Log.error("One and only one topic allowed in a TopicAndPartion map"); return false; } return true; }    }
@Test public void testMapOK() throws Exception { Map<TopicAndPartition, Long> tap = new HashMap<>(); tap.put(new TopicAndPartition(testTopic, 1), 1L); tap.put(new TopicAndPartition(testTopic, 2), 2L); assertTrue(KafkaOffsetManagerUtils.isValidOffsetMap(tap)); }
@Test public void testMapWithBelowZeroOffset() throws Exception { Map<TopicAndPartition, Long> tap = new HashMap<>(); tap.put(new TopicAndPartition(testTopic, 1), 1L); tap.put(new TopicAndPartition(testTopic, 2), -2L); assertFalse(KafkaOffsetManagerUtils.isValidOffsetMap(tap)); }
@Test public void testMapWithMultipleTopics() throws Exception { Map<TopicAndPartition, Long> tap = new HashMap<>(); tap.put(new TopicAndPartition(testTopic, 1), 1L); tap.put(new TopicAndPartition(testTopic + "1", 2), 2L); assertFalse(KafkaOffsetManagerUtils.isValidOffsetMap(tap)); }
@Test public void testMapWithPartitionBelowZero() throws Exception { Map<TopicAndPartition, Long> tap = new HashMap<>(); tap.put(new TopicAndPartition(testTopic, -1), 1L); assertFalse(KafkaOffsetManagerUtils.isValidOffsetMap(tap)); }