[ https://issues.apache.org/jira/browse/FLINK-4821?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15974155#comment-15974155 ]
ASF GitHub Bot commented on FLINK-4821: --------------------------------------- Github user tzulitai commented on a diff in the pull request: https://github.com/apache/flink/pull/3001#discussion_r112123170 --- Diff: flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java --- @@ -559,48 +699,298 @@ public void testFetcherShouldNotBeRestoringFromFailureIfNotRestoringFromCheckpoi @Test @SuppressWarnings("unchecked") + public void testFetcherShouldBeCorrectlySeededIfRestoringFromLegacyCheckpoint() throws Exception { + HashMap<KinesisStreamShard, SequenceNumber> fakeRestoredState = getFakeRestoredStore("all"); + + KinesisDataFetcher mockedFetcher = Mockito.mock(KinesisDataFetcher.class); + List<KinesisStreamShard> shards = new ArrayList<>(); + shards.addAll(fakeRestoredState.keySet()); + when(mockedFetcher.discoverNewShardsToSubscribe()).thenReturn(shards); + PowerMockito.whenNew(KinesisDataFetcher.class).withAnyArguments().thenReturn(mockedFetcher); + + // assume the given config is correct + PowerMockito.mockStatic(KinesisConfigUtil.class); + PowerMockito.doNothing().when(KinesisConfigUtil.class); + + TestableFlinkKinesisConsumer consumer = new TestableFlinkKinesisConsumer( + "fakeStream", new Properties(), 10, 2); + consumer.restoreState(fakeRestoredState); + consumer.open(new Configuration()); + consumer.run(Mockito.mock(SourceFunction.SourceContext.class)); + + Mockito.verify(mockedFetcher).setIsRestoringFromFailure(true); + for (Map.Entry<KinesisStreamShard, SequenceNumber> restoredShard : fakeRestoredState.entrySet()) { + Mockito.verify(mockedFetcher).advanceLastDiscoveredShardOfStream( + restoredShard.getKey().getStreamName(), restoredShard.getKey().getShard().getShardId()); + Mockito.verify(mockedFetcher).registerNewSubscribedShardState( + new KinesisStreamShardState(restoredShard.getKey(), restoredShard.getValue())); + } + } + + @Test + @SuppressWarnings("unchecked") public void testFetcherShouldBeCorrectlySeededIfRestoringFromCheckpoint() throws Exception { + // ---------------------------------------------------------------------- + // setting initial state + // ---------------------------------------------------------------------- + HashMap<KinesisStreamShard, SequenceNumber> fakeRestoredState = getFakeRestoredStore("all"); + + // ---------------------------------------------------------------------- + // mock operator state backend and initial state for initializeState() + // ---------------------------------------------------------------------- + TestingListState<Serializable> listState = new TestingListState<>(); + for (Map.Entry<KinesisStreamShard, SequenceNumber> state: fakeRestoredState.entrySet()) { + listState.add(Tuple2.of(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); + + // ---------------------------------------------------------------------- + // mock fetcher + // ---------------------------------------------------------------------- KinesisDataFetcher mockedFetcher = Mockito.mock(KinesisDataFetcher.class); + List<KinesisStreamShard> shards = new ArrayList<>(); + shards.addAll(fakeRestoredState.keySet()); + when(mockedFetcher.discoverNewShardsToSubscribe()).thenReturn(shards); PowerMockito.whenNew(KinesisDataFetcher.class).withAnyArguments().thenReturn(mockedFetcher); // assume the given config is correct PowerMockito.mockStatic(KinesisConfigUtil.class); PowerMockito.doNothing().when(KinesisConfigUtil.class); - HashMap<KinesisStreamShard, SequenceNumber> fakeRestoredState = new HashMap<>(); - fakeRestoredState.put( - new KinesisStreamShard("fakeStream1", - new Shard().withShardId(KinesisShardIdGenerator.generateFromShardOrder(0))), - new SequenceNumber(UUID.randomUUID().toString())); - fakeRestoredState.put( - new KinesisStreamShard("fakeStream1", - new Shard().withShardId(KinesisShardIdGenerator.generateFromShardOrder(1))), - new SequenceNumber(UUID.randomUUID().toString())); - fakeRestoredState.put( - new KinesisStreamShard("fakeStream1", - new Shard().withShardId(KinesisShardIdGenerator.generateFromShardOrder(2))), - new SequenceNumber(UUID.randomUUID().toString())); - fakeRestoredState.put( - new KinesisStreamShard("fakeStream2", - new Shard().withShardId(KinesisShardIdGenerator.generateFromShardOrder(0))), - new SequenceNumber(UUID.randomUUID().toString())); - fakeRestoredState.put( - new KinesisStreamShard("fakeStream2", - new Shard().withShardId(KinesisShardIdGenerator.generateFromShardOrder(1))), - new SequenceNumber(UUID.randomUUID().toString())); + // ---------------------------------------------------------------------- + // start to test seed initial state to fetcher + // ---------------------------------------------------------------------- + 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).setIsRestoringFromFailure(true); + for (Map.Entry<KinesisStreamShard, SequenceNumber> restoredShard : fakeRestoredState.entrySet()) { + Mockito.verify(mockedFetcher).advanceLastDiscoveredShardOfStream( + restoredShard.getKey().getStreamName(), restoredShard.getKey().getShard().getShardId()); + Mockito.verify(mockedFetcher).registerNewSubscribedShardState( + new KinesisStreamShardState(restoredShard.getKey(), restoredShard.getValue())); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testFetcherShouldBeCorrectlySeededOnlyItsOwnStates() throws Exception { + // ---------------------------------------------------------------------- + // setting initial state + // ---------------------------------------------------------------------- + HashMap<KinesisStreamShard, SequenceNumber> fakeRestoredState = getFakeRestoredStore("fakeStream1"); + + HashMap<KinesisStreamShard, SequenceNumber> fakeRestoredStateForOthers = getFakeRestoredStore("fakeStream2"); + + // ---------------------------------------------------------------------- + // mock operator state backend and initial state for initializeState() + // ---------------------------------------------------------------------- + TestingListState<Serializable> listState = new TestingListState<>(); + for (Map.Entry<KinesisStreamShard, SequenceNumber> state: fakeRestoredState.entrySet()) { + listState.add(Tuple2.of(state.getKey(), state.getValue())); + } + for (Map.Entry<KinesisStreamShard, SequenceNumber> state: fakeRestoredStateForOthers.entrySet()) { + listState.add(Tuple2.of(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); + + // ---------------------------------------------------------------------- + // mock fetcher + // ---------------------------------------------------------------------- + KinesisDataFetcher mockedFetcher = Mockito.mock(KinesisDataFetcher.class); + List<KinesisStreamShard> shards = new ArrayList<>(); + shards.addAll(fakeRestoredState.keySet()); + when(mockedFetcher.discoverNewShardsToSubscribe()).thenReturn(shards); + PowerMockito.whenNew(KinesisDataFetcher.class).withAnyArguments().thenReturn(mockedFetcher); + + // assume the given config is correct + PowerMockito.mockStatic(KinesisConfigUtil.class); + PowerMockito.doNothing().when(KinesisConfigUtil.class); + + // ---------------------------------------------------------------------- + // start to test seed initial state to fetcher + // ---------------------------------------------------------------------- TestableFlinkKinesisConsumer consumer = new TestableFlinkKinesisConsumer( "fakeStream", new Properties(), 10, 2); - consumer.restoreState(fakeRestoredState); + consumer.initializeState(initializationContext); consumer.open(new Configuration()); consumer.run(Mockito.mock(SourceFunction.SourceContext.class)); Mockito.verify(mockedFetcher).setIsRestoringFromFailure(true); + for (Map.Entry<KinesisStreamShard, SequenceNumber> restoredShard : fakeRestoredStateForOthers.entrySet()) { + // should never get restored state not belonging to itself + Mockito.verify(mockedFetcher, never()).advanceLastDiscoveredShardOfStream( + restoredShard.getKey().getStreamName(), restoredShard.getKey().getShard().getShardId()); + Mockito.verify(mockedFetcher, never()).registerNewSubscribedShardState( + new KinesisStreamShardState(restoredShard.getKey(), restoredShard.getValue())); + } for (Map.Entry<KinesisStreamShard, SequenceNumber> restoredShard : fakeRestoredState.entrySet()) { + // should get restored state belonging to itself Mockito.verify(mockedFetcher).advanceLastDiscoveredShardOfStream( restoredShard.getKey().getStreamName(), restoredShard.getKey().getShard().getShardId()); Mockito.verify(mockedFetcher).registerNewSubscribedShardState( new KinesisStreamShardState(restoredShard.getKey(), restoredShard.getValue())); } } + + @Test + @SuppressWarnings("unchecked") + /* + * If the original parallelism is 2 and states is: + * Consumer subtask 1: + * stream1, shard1, SequentialNumber(xxx) + * Consumer subtask 2: + * stream1, shard2, SequentialNumber(yyy) + * After discoverNewShardsToSubscribe() if there are two shards (shard3, shard4) been created: + * Consumer subtask 1 (late for discoverNewShardsToSubscribe()): + * stream1, shard1, SequentialNumber(xxx) + * Consumer subtask 2: + * stream1, shard2, SequentialNumber(yyy) + * stream1, shard4, SequentialNumber(zzz) + * If snapshotState() occur and parallelism is changed to 1: + * Broadcast state will be: --- End diff -- "Union" state > Implement rescalable non-partitioned state for Kinesis Connector > ---------------------------------------------------------------- > > Key: FLINK-4821 > URL: https://issues.apache.org/jira/browse/FLINK-4821 > Project: Flink > Issue Type: New Feature > Components: Kinesis Connector > Reporter: Tzu-Li (Gordon) Tai > Assignee: Wei-Che Wei > > FLINK-4379 added the rescalable non-partitioned state feature, along with the > implementation for the Kafka connector. > The AWS Kinesis connector will benefit from the feature and should implement > it too. This ticket tracks progress for this. -- This message was sent by Atlassian JIRA (v6.3.15#6346)