@Test public void testIsThisSubtaskShouldSubscribeTo() { assertTrue(KinesisDataFetcher.isThisSubtaskShouldSubscribeTo(0, 2, 0)); assertFalse(KinesisDataFetcher.isThisSubtaskShouldSubscribeTo(1, 2, 0)); assertTrue(KinesisDataFetcher.isThisSubtaskShouldSubscribeTo(2, 2, 0)); assertFalse(KinesisDataFetcher.isThisSubtaskShouldSubscribeTo(0, 2, 1)); assertTrue(KinesisDataFetcher.isThisSubtaskShouldSubscribeTo(1, 2, 1)); assertFalse(KinesisDataFetcher.isThisSubtaskShouldSubscribeTo(2, 2, 1)); }
public static boolean isThisSubtaskShouldSubscribeTo(int shardHash, int totalNumberOfConsumerSubtasks, int indexOfThisConsumerSubtask) { return (Math.abs(shardHash % totalNumberOfConsumerSubtasks)) == indexOfThisConsumerSubtask; }
KinesisDataFetcher { public static boolean isThisSubtaskShouldSubscribeTo(int shardHash, int totalNumberOfConsumerSubtasks, int indexOfThisConsumerSubtask) { return (Math.abs(shardHash % totalNumberOfConsumerSubtasks)) == indexOfThisConsumerSubtask; } }
KinesisDataFetcher { public static boolean isThisSubtaskShouldSubscribeTo(int shardHash, int totalNumberOfConsumerSubtasks, int indexOfThisConsumerSubtask) { return (Math.abs(shardHash % totalNumberOfConsumerSubtasks)) == indexOfThisConsumerSubtask; } KinesisDataFetcher(List<String> streams,
							SourceFunction.SourceContext<T> sourceContext,
							RuntimeContext runtimeContext,
							Properties configProps,
							KinesisDeserializationSchema<T> deserializationSchema,
							KinesisShardAssigner shardAssigner,
							AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
							WatermarkTracker watermarkTracker); @VisibleForTesting protected  KinesisDataFetcher(List<String> streams,
								SourceFunction.SourceContext<T> sourceContext,
								Object checkpointLock,
								RuntimeContext runtimeContext,
								Properties configProps,
								KinesisDeserializationSchema<T> deserializationSchema,
								KinesisShardAssigner shardAssigner,
								AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
								WatermarkTracker watermarkTracker,
								AtomicReference<Throwable> error,
								List<KinesisStreamShardState> subscribedShardsState,
								HashMap<String, String> subscribedStreamsToLastDiscoveredShardIds,
								FlinkKinesisProxyFactory kinesisProxyFactory); }
KinesisDataFetcher { public static boolean isThisSubtaskShouldSubscribeTo(int shardHash, int totalNumberOfConsumerSubtasks, int indexOfThisConsumerSubtask) { return (Math.abs(shardHash % totalNumberOfConsumerSubtasks)) == indexOfThisConsumerSubtask; } KinesisDataFetcher(List<String> streams,
							SourceFunction.SourceContext<T> sourceContext,
							RuntimeContext runtimeContext,
							Properties configProps,
							KinesisDeserializationSchema<T> deserializationSchema,
							KinesisShardAssigner shardAssigner,
							AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
							WatermarkTracker watermarkTracker); @VisibleForTesting protected  KinesisDataFetcher(List<String> streams,
								SourceFunction.SourceContext<T> sourceContext,
								Object checkpointLock,
								RuntimeContext runtimeContext,
								Properties configProps,
								KinesisDeserializationSchema<T> deserializationSchema,
								KinesisShardAssigner shardAssigner,
								AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
								WatermarkTracker watermarkTracker,
								AtomicReference<Throwable> error,
								List<KinesisStreamShardState> subscribedShardsState,
								HashMap<String, String> subscribedStreamsToLastDiscoveredShardIds,
								FlinkKinesisProxyFactory kinesisProxyFactory); void runFetcher(); HashMap<StreamShardMetadata, SequenceNumber> snapshotState(); void shutdownFetcher(); void awaitTermination(); void advanceLastDiscoveredShardOfStream(String stream, String shardId); List<StreamShardHandle> discoverNewShardsToSubscribe(); int registerNewSubscribedShardState(KinesisStreamShardState newSubscribedShardState); static boolean isThisSubtaskShouldSubscribeTo(int shardHash,
														int totalNumberOfConsumerSubtasks,
														int indexOfThisConsumerSubtask); @VisibleForTesting List<KinesisStreamShardState> getSubscribedShardsState(); static StreamShardMetadata convertToStreamShardMetadata(StreamShardHandle streamShardHandle); static StreamShardHandle convertToStreamShardHandle(StreamShardMetadata streamShardMetadata); }
KinesisDataFetcher { public static boolean isThisSubtaskShouldSubscribeTo(int shardHash, int totalNumberOfConsumerSubtasks, int indexOfThisConsumerSubtask) { return (Math.abs(shardHash % totalNumberOfConsumerSubtasks)) == indexOfThisConsumerSubtask; } KinesisDataFetcher(List<String> streams,
							SourceFunction.SourceContext<T> sourceContext,
							RuntimeContext runtimeContext,
							Properties configProps,
							KinesisDeserializationSchema<T> deserializationSchema,
							KinesisShardAssigner shardAssigner,
							AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
							WatermarkTracker watermarkTracker); @VisibleForTesting protected  KinesisDataFetcher(List<String> streams,
								SourceFunction.SourceContext<T> sourceContext,
								Object checkpointLock,
								RuntimeContext runtimeContext,
								Properties configProps,
								KinesisDeserializationSchema<T> deserializationSchema,
								KinesisShardAssigner shardAssigner,
								AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
								WatermarkTracker watermarkTracker,
								AtomicReference<Throwable> error,
								List<KinesisStreamShardState> subscribedShardsState,
								HashMap<String, String> subscribedStreamsToLastDiscoveredShardIds,
								FlinkKinesisProxyFactory kinesisProxyFactory); void runFetcher(); HashMap<StreamShardMetadata, SequenceNumber> snapshotState(); void shutdownFetcher(); void awaitTermination(); void advanceLastDiscoveredShardOfStream(String stream, String shardId); List<StreamShardHandle> discoverNewShardsToSubscribe(); int registerNewSubscribedShardState(KinesisStreamShardState newSubscribedShardState); static boolean isThisSubtaskShouldSubscribeTo(int shardHash,
														int totalNumberOfConsumerSubtasks,
														int indexOfThisConsumerSubtask); @VisibleForTesting List<KinesisStreamShardState> getSubscribedShardsState(); static StreamShardMetadata convertToStreamShardMetadata(StreamShardHandle streamShardHandle); static StreamShardHandle convertToStreamShardHandle(StreamShardMetadata streamShardMetadata); static final KinesisShardAssigner DEFAULT_SHARD_ASSIGNER; }