This repository has been archived by the owner on Jun 7, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 292
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #797 from zalando/ARUHA-1330
ARUHA-1330 Fix nakadi-cursors comparison and usage of finished timelines
- Loading branch information
Showing
23 changed files
with
401 additions
and
84 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
90 changes: 90 additions & 0 deletions
90
src/main/java/org/zalando/nakadi/service/NakadiCursorComparator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| package org.zalando.nakadi.service; | ||
|
|
||
| import org.zalando.nakadi.domain.NakadiCursor; | ||
| import org.zalando.nakadi.domain.Timeline; | ||
| import org.zalando.nakadi.exceptions.InternalNakadiException; | ||
| import org.zalando.nakadi.exceptions.NoSuchEventTypeException; | ||
| import org.zalando.nakadi.repository.db.EventTypeCache; | ||
|
|
||
| import java.util.Comparator; | ||
| import java.util.Iterator; | ||
| import java.util.List; | ||
| import java.util.Objects; | ||
|
|
||
| public class NakadiCursorComparator implements Comparator<NakadiCursor> { | ||
| private final EventTypeCache eventTypeCache; | ||
|
|
||
| public NakadiCursorComparator(final EventTypeCache eventTypeCache) { | ||
| this.eventTypeCache = eventTypeCache; | ||
| } | ||
|
|
||
| public int compare(final NakadiCursor c1, final NakadiCursor c2) { | ||
| if (!Objects.equals(c1.getEventType(), c2.getEventType())) { | ||
| throw new IllegalArgumentException("Cursors from different event types are not comparable"); | ||
| } | ||
| if (c1.getTimeline().getOrder() == c2.getTimeline().getOrder()) { | ||
| return c1.getOffset().compareTo(c2.getOffset()); | ||
| } | ||
| if (c1.getTimeline().getOrder() > c2.getTimeline().getOrder()) { | ||
| return -compareOrdered(c2, c1); | ||
| } else { | ||
| return compareOrdered(c1, c2); | ||
| } | ||
| } | ||
|
|
||
| private int compareOrdered(final NakadiCursor c1, final NakadiCursor c2) { | ||
| // Disclaimer: The reason of this method complexity is to avoid objects creation. | ||
|
|
||
| // If c2 moved from -1, than it is definitely greater. | ||
| if (!StaticStorageWorkerFactory.get(c2.getTimeline().getStorage()).isInitialOffset(c2.getOffset())) { | ||
| return -1; | ||
| } | ||
|
|
||
| Iterator<Timeline> timelineIterator = null; | ||
|
|
||
| NakadiCursor first = c1; | ||
| // Handle obsolete timeline information | ||
| if (first.getTimeline().getLatestPosition() == null) { | ||
| timelineIterator = createTimelinesIterator(first.getEventType(), first.getTimeline().getOrder()); | ||
| first = new NakadiCursor(timelineIterator.next(), first.getPartition(), first.getOffset()); | ||
| } | ||
|
|
||
| while (first.getTimeline().getOrder() != c2.getTimeline().getOrder()) { | ||
| final boolean isFirstAtEndOfTimeline = StaticStorageWorkerFactory.get(first.getTimeline().getStorage()) | ||
| .isLastOffsetForPartition(first.getTimeline(), first.getPartition(), first.getOffset()); | ||
| if (!isFirstAtEndOfTimeline) { | ||
| return -1; | ||
| } | ||
| if (null == timelineIterator) { | ||
| timelineIterator = createTimelinesIterator(first.getEventType(), first.getTimeline().getOrder() + 1); | ||
| } | ||
| final Timeline nextTimeline = timelineIterator.next(); | ||
| final String initialOffset = StaticStorageWorkerFactory.get(nextTimeline.getStorage()) | ||
| .getFirstOffsetInTimeline(first.getPartition()); | ||
| first = new NakadiCursor( | ||
| nextTimeline, | ||
| first.getPartition(), | ||
| initialOffset); | ||
| } | ||
| return first.getOffset().compareTo(c2.getOffset()); | ||
| } | ||
|
|
||
|
|
||
| private Iterator<Timeline> createTimelinesIterator(final String eventType, final int order) { | ||
| try { | ||
| final List<Timeline> timelines = eventTypeCache.getTimelinesOrdered(eventType); | ||
| final Iterator<Timeline> result = timelines.iterator(); | ||
| if (timelines.get(0).getOrder() == order) { | ||
| return result; | ||
| } | ||
| // I do not want to handle hasNext | ||
| while (result.next().getOrder() != (order - 1)) { | ||
| } | ||
| return result; | ||
| } catch (final NoSuchEventTypeException | InternalNakadiException ex) { | ||
| // The reason for runtime exception is that cursors are constructed before, and probably should be working. | ||
| // Otherwise it makes no sense for this exception. | ||
| throw new IllegalStateException(ex); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.