flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
/**
	 * FLINK-8484: ensure that a state change in the StreamShardMetadata other than {@link StreamShardMetadata#shardId} or
	 * {@link StreamShardMetadata#streamName} does not result in the shard not being able to be restored.
	 * This handles the corner case where the stored shard metadata is open (no ending sequence number), but after the
	 * job restore, the shard has been closed (ending number set) due to re-sharding, and we can no longer rely on
	 * {@link StreamShardMetadata#equals(Object)} to find back the sequence number in the collection of restored shard metadata.
	 * <p></p>
	 * Therefore, we will rely on synchronizing the snapshot's state with the Kinesis shard before attempting to find back
	 * the sequence number to restore.
	 */
FLINK-8484: ensure that a state change in the StreamShardMetadata other than  StreamShardMetadata#shardId or
StreamShardMetadata#streamName does not result in the shard not being able to be restored.
This handles the corner case where the stored shard metadata is open (no ending sequence number), but after the
job restore, the shard has been closed (ending number set) due to re-sharding, and we can no longer rely on
StreamShardMetadata#equals(Object) to find back the sequence number in the collection of restored shard metadata.

Therefore, we will rely on synchronizing the snapshot's state with the Kinesis shard before attempting to find back
the sequence number to restore.
FLINK - 8484 : ensure that a state change in the StreamShardMetadata other than StreamShardMetadata#shardId or StreamShardMetadata#streamName does not result in the shard not being able to be restored . This handles the corner case where the stored shard metadata is open ( no ending sequence number ) but after the job restore the shard has been closed ( ending number set ) due to re - sharding and we can no longer rely on StreamShardMetadata#equals ( Object ) to find back the sequence number in the collection of restored shard metadata . Therefore we will rely on synchronizing the snapshot ' s state with the Kinesis shard before attempting to find back the sequence number to restore .
@Test
	public void testFindSequenceNumberToRestoreFromIfTheShardHasBeenClosedSinceTheStateWasStored() throws Exception {
		HashMap<StreamShardHandle, SequenceNumber> fakeRestoredState = getFakeRestoredStore("all");
		TestingListState<Tuple2<StreamShardMetadata, SequenceNumber>> listState = new TestingListState<>();
		for (Map.Entry<StreamShardHandle, SequenceNumber> state : fakeRestoredState.entrySet()) {
			listState.add(Tuple2.of(KinesisDataFetcher.convertToStreamShardMetadata(state.getKey()), state.getValue()));
		}
		OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
		when(operatorStateStore.getUnionListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
		StateInitializationContext initializationContext = mock(StateInitializationContext.class);
		when(initializationContext.getOperatorStateStore()).thenReturn(operatorStateStore);
		when(initializationContext.isRestored()).thenReturn(true);
		KinesisDataFetcher mockedFetcher = Mockito.mock(KinesisDataFetcher.class);
		List<StreamShardHandle> shards = new ArrayList<>();
		final StreamShardHandle originalStreamShardHandle = fakeRestoredState.keySet().iterator().next();
		final StreamShardHandle closedStreamShardHandle = new StreamShardHandle(originalStreamShardHandle.getStreamName(), originalStreamShardHandle.getShard());
		final SequenceNumberRange sequenceNumberRange = new SequenceNumberRange();
		sequenceNumberRange.setEndingSequenceNumber("1293844");
		closedStreamShardHandle.getShard().setSequenceNumberRange(sequenceNumberRange);
		shards.add(closedStreamShardHandle);
		when(mockedFetcher.discoverNewShardsToSubscribe()).thenReturn(shards);
		PowerMockito.whenNew(KinesisDataFetcher.class).withAnyArguments().thenReturn(mockedFetcher);
		PowerMockito.mockStatic(KinesisConfigUtil.class);
		PowerMockito.doNothing().when(KinesisConfigUtil.class);
		TestableFlinkKinesisConsumer consumer = new TestableFlinkKinesisConsumer(
			"fakeStream", new Properties(), 10, 2);
		consumer.initializeState(initializationContext);
		consumer.open(new Configuration());
		consumer.run(Mockito.mock(SourceFunction.SourceContext.class));
		Mockito.verify(mockedFetcher).registerNewSubscribedShardState(
			new KinesisStreamShardState(KinesisDataFetcher.convertToStreamShardMetadata(closedStreamShardHandle),
				closedStreamShardHandle, fakeRestoredState.get(closedStreamShardHandle)));
	}
@ Test public void testFindSequenceNumberToRestoreFromIfTheShardHasBeenClosedSinceTheStateWasStored ( ) throws Exception { HashMap < StreamShardHandle , SequenceNumber > fakeRestoredState = getFakeRestoredStore ( "all" ) ; TestingListState < Tuple2 < StreamShardMetadata , SequenceNumber > > listState = new TestingListState < > ( ) ; for ( Map . Entry < StreamShardHandle , SequenceNumber > state : fakeRestoredState . entrySet ( ) ) { listState . add ( Tuple2 . of ( KinesisDataFetcher . convertToStreamShardMetadata ( state . getKey ( ) ) , state . getValue ( ) ) ) ; } OperatorStateStore operatorStateStore = mock ( OperatorStateStore . class ) ; when ( operatorStateStore . getUnionListState ( Matchers . any ( ListStateDescriptor . class ) ) ) . thenReturn ( listState ) ; StateInitializationContext initializationContext = mock ( StateInitializationContext . class ) ; when ( initializationContext . getOperatorStateStore ( ) ) . thenReturn ( operatorStateStore ) ; when ( initializationContext . isRestored ( ) ) . thenReturn ( true ) ; KinesisDataFetcher mockedFetcher = Mockito . mock ( KinesisDataFetcher . class ) ; List < StreamShardHandle > shards = new ArrayList < > ( ) ; final StreamShardHandle originalStreamShardHandle = fakeRestoredState . keySet ( ) . iterator ( ) . next ( ) ; final StreamShardHandle closedStreamShardHandle = new StreamShardHandle ( originalStreamShardHandle . getStreamName ( ) , originalStreamShardHandle . getShard ( ) ) ; final SequenceNumberRange sequenceNumberRange = new SequenceNumberRange ( ) ; sequenceNumberRange . setEndingSequenceNumber ( "1293844" ) ; closedStreamShardHandle . getShard ( ) . setSequenceNumberRange ( sequenceNumberRange ) ; shards . add ( closedStreamShardHandle ) ; when ( mockedFetcher . discoverNewShardsToSubscribe ( ) ) . thenReturn ( shards ) ; PowerMockito . whenNew ( KinesisDataFetcher . class ) . withAnyArguments ( ) . thenReturn ( mockedFetcher ) ; PowerMockito . mockStatic ( KinesisConfigUtil . class ) ; PowerMockito . doNothing ( ) . when ( KinesisConfigUtil . class ) ; TestableFlinkKinesisConsumer consumer = new TestableFlinkKinesisConsumer ( "fakeStream" , new Properties ( ) , 10 , 2 ) ; consumer . initializeState ( initializationContext ) ; consumer . open ( new Configuration ( ) ) ; consumer . run ( Mockito . mock ( SourceFunction . SourceContext . class ) ) ; Mockito . verify ( mockedFetcher ) . registerNewSubscribedShardState ( new KinesisStreamShardState ( KinesisDataFetcher . convertToStreamShardMetadata ( closedStreamShardHandle ) , closedStreamShardHandle , fakeRestoredState . get ( closedStreamShardHandle ) ) ) ; }
FLINK-8484: ensure that a state change in the StreamShardMetadata other than {@link StreamShardMetadata#shardId} or
 {@link StreamShardMetadata#streamName} does not result in the shard not being able to be restored.
FLINK - 8484 : ensure that a state change in the StreamShardMetadata other than { @link StreamShardMetadata#shardId } or { @link StreamShardMetadata#streamName } does not result in the shard not being able to be restored .
// ---------------------------------------------------------------------- // setup initial state // ---------------------------------------------------------------------- // ---------------------------------------------------------------------- // mock operator state backend and initial state for initializeState() // ---------------------------------------------------------------------- // ---------------------------------------------------------------------- // mock fetcher // ---------------------------------------------------------------------- // create a fake stream shard handle based on the first entry in the restored state // close the shard handle by setting an ending sequence number // assume the given config is correct // ---------------------------------------------------------------------- // start to test fetcher's initial state seeding // ----------------------------------------------------------------------