flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-4821) Implement rescalable non-partitioned state for Kinesis Connector
Date Wed, 19 Apr 2017 06:29:42 GMT

    [ https://issues.apache.org/jira/browse/FLINK-4821?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15974158#comment-15974158
] 

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_r112122730
  
    --- Diff: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumer.java
---
    @@ -66,31 +80,48 @@
     	//  Consumer properties
     	// ------------------------------------------------------------------------
     
    -	/** The names of the Kinesis streams that we will be consuming from */
    +	/**
    +	 * The names of the Kinesis streams that we will be consuming from
    +	 */
     	private final List<String> streams;
     
    -	/** Properties to parametrize settings such as AWS service region, initial position
in stream,
    -	 * shard list retrieval behaviours, etc */
    +	/**
    +	 * Properties to parametrize settings such as AWS service region, initial position in
stream,
    +	 * shard list retrieval behaviours, etc
    +	 */
     	private final Properties configProps;
     
    -	/** User supplied deseriliazation schema to convert Kinesis byte messages to Flink objects
*/
    +	/**
    +	 * User supplied deseriliazation schema to convert Kinesis byte messages to Flink objects
    +	 */
     	private final KinesisDeserializationSchema<T> deserializer;
     
     	// ------------------------------------------------------------------------
     	//  Runtime state
     	// ------------------------------------------------------------------------
     
    -	/** Per-task fetcher for Kinesis data records, where each fetcher pulls data from one
or more Kinesis shards */
    +	/**
    +	 * Per-task fetcher for Kinesis data records, where each fetcher pulls data from one
or more Kinesis shards
    +	 */
     	private transient KinesisDataFetcher<T> fetcher;
     
    -	/** The sequence numbers in the last state snapshot of this subtask */
    +	/**
    +	 * The sequence numbers in the last state snapshot of this subtask
    +	 */
     	private transient HashMap<KinesisStreamShard, SequenceNumber> lastStateSnapshot;
     
    -	/** The sequence numbers to restore to upon restore from failure */
    +	/**
    +	 * The sequence numbers to restore to upon restore from failure
    +	 */
     	private transient HashMap<KinesisStreamShard, SequenceNumber> sequenceNumsToRestore;
     
     	private volatile boolean running = true;
     
    +	// ------------------------------------------------------------------------
    +	// State for Checkpoint
    +	// ------------------------------------------------------------------------
    +
    +	private transient ListState<Tuple2<KinesisStreamShard, SequenceNumber>>
offsetsStateForCheckpoint;
    --- End diff --
    
    "offset" is the Kafka term.
    I would try to rename this to use "sequence number" instead (or a likewise abbreviation).


> 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)

Mime
View raw message