StartingPointShardsFinder implements Serializable { Set<Shard> findShardsAtStartingPoint( SimplifiedKinesisClient kinesis, String streamName, StartingPoint startingPoint) throws TransientKinesisException { List<Shard> allShards = kinesis.listShards(streamName); Set<Shard> initialShards = findInitialShardsWithoutParents(streamName, allShards); Set<Shard> startingPointShards = new HashSet<>(); Set<Shard> expiredShards; do { Set<Shard> validShards = validateShards(kinesis, initialShards, streamName, startingPoint); startingPointShards.addAll(validShards); expiredShards = Sets.difference(initialShards, validShards); if (!expiredShards.isEmpty()) { LOG.info( "Following shards expired for {} stream at '{}' starting point: {}", streamName, startingPoint, expiredShards); } initialShards = findNextShards(allShards, expiredShards); } while (!expiredShards.isEmpty()); return startingPointShards; }    }
@Test public void shouldFindFirstShardsWhenAllShardsAreValid() throws Exception { Instant timestampAtTheBeginning = new Instant(); StartingPoint startingPointAtTheBeginning = new StartingPoint(timestampAtTheBeginning); for (Shard shard : allShards) { activeAtTimestamp(shard, timestampAtTheBeginning); } when(kinesis.listShards(STREAM_NAME)).thenReturn(allShards); Iterable<Shard> shardsAtStartingPoint = underTest.findShardsAtStartingPoint(kinesis, STREAM_NAME, startingPointAtTheBeginning); assertThat(shardsAtStartingPoint).containsExactlyInAnyOrder(shard00, shard01); }
@Test public void shouldFind3StartingShardsInTheMiddle() throws Exception { Instant timestampAfterShards3And4Merge = new Instant(); StartingPoint startingPointAfterFirstSplitsAndMerge = new StartingPoint(timestampAfterShards3And4Merge); expiredAtTimestamp(shard00, timestampAfterShards3And4Merge); expiredAtTimestamp(shard01, timestampAfterShards3And4Merge); activeAtTimestamp(shard02, timestampAfterShards3And4Merge); expiredAtTimestamp(shard03, timestampAfterShards3And4Merge); expiredAtTimestamp(shard04, timestampAfterShards3And4Merge); activeAtTimestamp(shard05, timestampAfterShards3And4Merge); activeAtTimestamp(shard06, timestampAfterShards3And4Merge); activeAtTimestamp(shard07, timestampAfterShards3And4Merge); activeAtTimestamp(shard08, timestampAfterShards3And4Merge); activeAtTimestamp(shard09, timestampAfterShards3And4Merge); activeAtTimestamp(shard10, timestampAfterShards3And4Merge); when(kinesis.listShards(STREAM_NAME)).thenReturn(allShards); Iterable<Shard> shardsAtStartingPoint = underTest.findShardsAtStartingPoint( kinesis, STREAM_NAME, startingPointAfterFirstSplitsAndMerge); assertThat(shardsAtStartingPoint).containsExactlyInAnyOrder(shard02, shard05, shard06); }
@Test public void shouldFindLastShardWhenAllPreviousExpired() throws Exception { Instant timestampAtTheEnd = new Instant(); StartingPoint startingPointAtTheEnd = new StartingPoint(timestampAtTheEnd); expiredAtTimestamp(shard00, timestampAtTheEnd); expiredAtTimestamp(shard01, timestampAtTheEnd); expiredAtTimestamp(shard02, timestampAtTheEnd); expiredAtTimestamp(shard03, timestampAtTheEnd); expiredAtTimestamp(shard04, timestampAtTheEnd); expiredAtTimestamp(shard05, timestampAtTheEnd); expiredAtTimestamp(shard06, timestampAtTheEnd); expiredAtTimestamp(shard07, timestampAtTheEnd); expiredAtTimestamp(shard08, timestampAtTheEnd); activeAtTimestamp(shard09, timestampAtTheEnd); activeAtTimestamp(shard10, timestampAtTheEnd); when(kinesis.listShards(STREAM_NAME)).thenReturn(allShards); Iterable<Shard> shardsAtStartingPoint = underTest.findShardsAtStartingPoint(kinesis, STREAM_NAME, startingPointAtTheEnd); assertThat(shardsAtStartingPoint).containsExactlyInAnyOrder(shard09, shard10); }
@Test public void shouldFindLastShardsWhenLatestStartingPointRequested() throws Exception { StartingPoint latestStartingPoint = new StartingPoint(InitialPositionInStream.LATEST); when(kinesis.listShards(STREAM_NAME)).thenReturn(allShards); Iterable<Shard> shardsAtStartingPoint = underTest.findShardsAtStartingPoint(kinesis, STREAM_NAME, latestStartingPoint); assertThat(shardsAtStartingPoint).containsExactlyInAnyOrder(shard09, shard10); }
@Test public void shouldFindEarliestShardsWhenTrimHorizonStartingPointRequested() throws Exception { StartingPoint trimHorizonStartingPoint = new StartingPoint(InitialPositionInStream.TRIM_HORIZON); when(kinesis.listShards(STREAM_NAME)).thenReturn(allShards); Iterable<Shard> shardsAtStartingPoint = underTest.findShardsAtStartingPoint(kinesis, STREAM_NAME, trimHorizonStartingPoint); assertThat(shardsAtStartingPoint).containsExactlyInAnyOrder(shard00, shard01); }
@Test(expected = IllegalStateException.class) public void shouldThrowExceptionWhenSuccessorsNotFoundForExpiredShard() throws Exception { StartingPoint latestStartingPoint = new StartingPoint(InitialPositionInStream.LATEST); Shard closedShard10 = createClosedShard("0010") .toBuilder() .parentShardId("0008") .adjacentParentShardId("0005") .build(); List<Shard> shards = ImmutableList.of( shard00, shard01, shard02, shard03, shard04, shard05, shard06, shard07, shard08, shard09, closedShard10); when(kinesis.listShards(STREAM_NAME)).thenReturn(shards); underTest.findShardsAtStartingPoint(kinesis, STREAM_NAME, latestStartingPoint); }
@Test(expected = IllegalStateException.class) public void shouldThrowExceptionWhenSuccessorsNotFoundForExpiredShard() throws Exception { StartingPoint latestStartingPoint = new StartingPoint(InitialPositionInStream.LATEST); Shard closedShard10 = createClosedShard("0010").withParentShardId("0008").withAdjacentParentShardId("0005"); List<Shard> shards = ImmutableList.of( shard00, shard01, shard02, shard03, shard04, shard05, shard06, shard07, shard08, shard09, closedShard10); when(kinesis.listShards(STREAM_NAME)).thenReturn(shards); underTest.findShardsAtStartingPoint(kinesis, STREAM_NAME, latestStartingPoint); }