Github user skidder commented on the issue:
https://github.com/apache/flink/pull/3458
@tzulitai I was able to scale up the number of shards on a Kinesis stream
without any interruption in processing. Looks good!
# Configuration
## Flink Cluster
Single Job Manager running as standalone cluster, with a single Task
Manager with 4 slots. Both Flink servers were built from source of this feature
branch.
## Application
Single Flink application running with parallelism of 4
## Kinesis stream
Stream name `mux_video_events_staging` with one shard
# Test Steps & Results
On startup the Flink application has Sub-task (1) read from the one shard
on the stream:
```
2017-03-29 15:55:03,306 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 1 will be seeded with initial shard
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000000,HashKeyRange: {StartingHashKey: 0,EndingHashKey:
340282366920938463463374607431768211455},SequenceNumberRange:
{StartingSequenceNumber:
49569797317567661038287361310393874557410775187880673282,}}'}, starting state
set as sequence number LATEST_SEQUENCE_NUM
2017-03-29 15:55:03,312 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 1 will start consuming seeded shard
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000000,HashKeyRange: {StartingHashKey: 0,EndingHashKey:
340282366920938463463374607431768211455},SequenceNumberRange:
{StartingSequenceNumber:
49569797317567661038287361310393874557410775187880673282,}}'} from sequence
number LATEST_SEQUENCE_NUM with ShardConsumer 0
```
Next, I increased the number of shards from 1 to 2. Sub-task (1) previously
responsible for reading from the one shard is marked as temporarily idle;
Sub-tasks (2) & (3) begin reading from the 2 new shards:
```
2017-03-29 17:36:36,741 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 3 has discovered a new shard
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000002,ParentShardId: shardId-000000000000,HashKeyRange:
{StartingHashKey: 170141183460469231731687303715884105729,EndingHashKey:
340282366920938463463374607431768211455},SequenceNumberRange:
{StartingSequenceNumber:
49571795731277369408532401914933651574587478731630051362,}}'} due to
resharding, and will start consuming the shard from sequence number
EARLIEST_SEQUENCE_NUM with ShardConsumer 0
2017-03-29 17:36:38,606 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 1 has reached the end of subscribed shard:
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000000,HashKeyRange: {StartingHashKey: 0,EndingHashKey:
340282366920938463463374607431768211455},SequenceNumberRange:
{StartingSequenceNumber:
49569797317567661038287361310393874557410775187880673282,}}'}
2017-03-29 17:36:38,606 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 1 has reached the end of all currently subscribed shards; marking the
subtask as temporarily idle ...
2017-03-29 17:36:45,240 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 2 has discovered a new shard
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000001,ParentShardId: shardId-000000000000,HashKeyRange:
{StartingHashKey: 0,EndingHashKey:
170141183460469231731687303715884105728},SequenceNumberRange:
{StartingSequenceNumber:
49571795731255068663333871291792115856314830370124070930,}}'} due to
resharding, and will start consuming the shard from sequence number
EARLIEST_SEQUENCE_NUM with ShardConsumer 0
```
I then increased the number of shards from 2 to 4. Sub-task (0) reads from
a new shard; Sub-tasks
(2) & (3) stop reading from their closed shards and begin reading from the
new shards; Sub-task (1), which was previously marked as temporarily-idle,
starts reading from a new shard:
```
2017-03-29 17:41:58,005 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 2 has reached the end of subscribed shard:
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000001,ParentShardId: shardId-000000000000,HashKeyRange:
{StartingHashKey: 0,EndingHashKey:
170141183460469231731687303715884105728},SequenceNumberRange:
{StartingSequenceNumber:
49571795731255068663333871291792115856314830370124070930,}}'}
2017-03-29 17:41:58,393 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 1 has discovered a new shard
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000004,ParentShardId: shardId-000000000001,HashKeyRange:
{StartingHashKey: 85070591730234615865843651857942052865,EndingHashKey:
170141183460469231731687303715884105728},SequenceNumberRange:
{StartingSequenceNumber:
49571795845144974392229763675615029074730034502679134274,}}'} due to
resharding, and will start consuming the shard from sequence number
EARLIEST_SEQUENCE_NUM with ShardConsumer 1
2017-03-29 17:41:59,602 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 0 has discovered a new shard
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000003,ParentShardId: shardId-000000000001,HashKeyRange:
{StartingHashKey: 0,EndingHashKey:
85070591730234615865843651857942052864},SequenceNumberRange:
{StartingSequenceNumber:
49571795845122673647031233052473493356457386141173153842,}}'} due to
resharding, and will start consuming the shard from sequence number
EARLIEST_SEQUENCE_NUM with ShardConsumer 0
2017-03-29 17:42:28,173 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 3 has reached the end of subscribed shard:
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000002,ParentShardId: shardId-000000000000,HashKeyRange:
{StartingHashKey: 170141183460469231731687303715884105729,EndingHashKey:
340282366920938463463374607431768211455},SequenceNumberRange:
{StartingSequenceNumber:
49571795731277369408532401914933651574587478731630051362,}}'}
2017-03-29 17:42:29,520 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 2 has discovered a new shard
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000005,ParentShardId: shardId-000000000002,HashKeyRange:
{StartingHashKey: 170141183460469231731687303715884105729,EndingHashKey:
255211775190703847597530955573826158593},SequenceNumberRange:
{StartingSequenceNumber:
49571795855871632832722993406693709563873898448640016466,}}'} due to
resharding, and will start consuming the shard from sequence number
EARLIEST_SEQUENCE_NUM with ShardConsumer 1
2017-03-29 17:42:31,183 INFO
org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
Subtask 3 has discovered a new shard
KinesisStreamShard{streamName='mux_video_events_staging', shard='{ShardId:
shardId-000000000006,ParentShardId: shardId-000000000002,HashKeyRange:
{StartingHashKey: 255211775190703847597530955573826158594,EndingHashKey:
340282366920938463463374607431768211455},SequenceNumberRange:
{StartingSequenceNumber:
49571795855893933577921524029835245282146546810145996898,}}'} due to
resharding, and will start consuming the shard from sequence number
EARLIEST_SEQUENCE_NUM with ShardConsumer 1
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
