@Test void testLowerBound() { Map<TopicPartition, Long> firstPositions = new HashMap<>(); firstPositions.put(new TopicPartition(TEST_TOPIC, 0), 0L); firstPositions.put(new TopicPartition(TEST_TOPIC, 1), 10L); firstPositions.put(new TopicPartition(TEST_TOPIC, 2), 2L); firstPositions.put(new TopicPartition(TEST_TOPIC, 3), 2L); KafkaTrackingToken first = newInstance(firstPositions); Map<TopicPartition, Long> secondPositions = new HashMap<>(); secondPositions.put(new TopicPartition(TEST_TOPIC, 0), 10L); secondPositions.put(new TopicPartition(TEST_TOPIC, 1), 1L); secondPositions.put(new TopicPartition(TEST_TOPIC, 2), 2L); secondPositions.put(new TopicPartition(TEST_TOPIC, 4), 3L); KafkaTrackingToken second = newInstance(secondPositions); Map<TopicPartition, Long> expectedPositions = new HashMap<>(); expectedPositions.put(new TopicPartition(TEST_TOPIC, 0), 0L); expectedPositions.put(new TopicPartition(TEST_TOPIC, 1), 1L); expectedPositions.put(new TopicPartition(TEST_TOPIC, 2), 2L); expectedPositions.put(new TopicPartition(TEST_TOPIC, 3), 0L); expectedPositions.put(new TopicPartition(TEST_TOPIC, 4), 0L); KafkaTrackingToken expected = newInstance(expectedPositions); assertEquals(expected, first.lowerBound(second)); }
@Override public TrackingToken lowerBound(TrackingToken other) { isTrue(other instanceof KafkaTrackingToken, () -> "Incompatible token type provided."); return new KafkaTrackingToken(bounds((KafkaTrackingToken) other, Math::min)); }
KafkaTrackingToken implements TrackingToken, Serializable { @Override public TrackingToken lowerBound(TrackingToken other) { isTrue(other instanceof KafkaTrackingToken, () -> "Incompatible token type provided."); return new KafkaTrackingToken(bounds((KafkaTrackingToken) other, Math::min)); } }
KafkaTrackingToken implements TrackingToken, Serializable { @Override public TrackingToken lowerBound(TrackingToken other) { isTrue(other instanceof KafkaTrackingToken, () -> "Incompatible token type provided."); return new KafkaTrackingToken(bounds((KafkaTrackingToken) other, Math::min)); } private  KafkaTrackingToken(Map<TopicPartition, Long> positions); }
KafkaTrackingToken implements TrackingToken, Serializable { @Override public TrackingToken lowerBound(TrackingToken other) { isTrue(other instanceof KafkaTrackingToken, () -> "Incompatible token type provided."); return new KafkaTrackingToken(bounds((KafkaTrackingToken) other, Math::min)); } private  KafkaTrackingToken(Map<TopicPartition, Long> positions); @JsonCreator static KafkaTrackingToken newInstance(@JsonProperty("positions") Map<TopicPartition, Long> positions); static KafkaTrackingToken emptyToken(); static boolean isNotEmpty(KafkaTrackingToken token); static boolean isEmpty(KafkaTrackingToken token); @SuppressWarnings("ConstantConditions") static KafkaTrackingToken from(TrackingToken trackingToken); Map<TopicPartition, Long> positions(); KafkaTrackingToken advancedTo(String topic, int partition, long offset); @Override TrackingToken lowerBound(TrackingToken other); @Override TrackingToken upperBound(TrackingToken other); @Override boolean covers(TrackingToken other); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString(); }
KafkaTrackingToken implements TrackingToken, Serializable { @Override public TrackingToken lowerBound(TrackingToken other) { isTrue(other instanceof KafkaTrackingToken, () -> "Incompatible token type provided."); return new KafkaTrackingToken(bounds((KafkaTrackingToken) other, Math::min)); } private  KafkaTrackingToken(Map<TopicPartition, Long> positions); @JsonCreator static KafkaTrackingToken newInstance(@JsonProperty("positions") Map<TopicPartition, Long> positions); static KafkaTrackingToken emptyToken(); static boolean isNotEmpty(KafkaTrackingToken token); static boolean isEmpty(KafkaTrackingToken token); @SuppressWarnings("ConstantConditions") static KafkaTrackingToken from(TrackingToken trackingToken); Map<TopicPartition, Long> positions(); KafkaTrackingToken advancedTo(String topic, int partition, long offset); @Override TrackingToken lowerBound(TrackingToken other); @Override TrackingToken upperBound(TrackingToken other); @Override boolean covers(TrackingToken other); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString();  }