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-5020) Make the GenericWriteAheadSink rescalable.
Date Mon, 07 Nov 2016 11:18:58 GMT

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

ASF GitHub Bot commented on FLINK-5020:
---------------------------------------

Github user zentol commented on a diff in the pull request:

    https://github.com/apache/flink/pull/2759#discussion_r86755620
  
    --- Diff: flink-streaming-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
---
    @@ -318,6 +312,46 @@ protected void verifyResultsDataDiscardingUponRestore(
     		Assert.assertTrue("The following ID's were not found in the ResultSet: " + list.toString(),
list.isEmpty());
     	}
     
    +	@Override
    +	protected void verifyResultsWhenScalingDown(CassandraTupleWriteAheadSink<Tuple3<String,
Integer, Integer>> sink) throws Exception {
    +
    +		ArrayList<Integer> list = new ArrayList<>();
    +		for (int x = 1; x < 34; x++) {
    +			list.add(x);
    +		}
    +
    +		ResultSet result = session.execute(SELECT_DATA_QUERY);
    +		for (Row s : result) {
    +			list.remove(new Integer(s.getInt("counter")));
    +		}
    +		Assert.assertTrue(list.isEmpty());
    +	}
    +
    +	@Override
    +	protected void verifyResultsWhenScalingUp(
    +		CassandraTupleWriteAheadSink<Tuple3<String, Integer, Integer>> sink, int
startElementCounter, int endElementCounter) {
    +
    +		// IMPORTANT NOTE:
    +		//
    +		// for cassandra we always have to start from 1 because
    +		// all operators will share the same final db
    +
    +		ArrayList<Integer> list = new ArrayList<>();
    +		for (int i = 1; i <= endElementCounter; i++) {
    +			list.add(i);
    +		}
    +
    +		ArrayList<Integer> expected = new ArrayList<>();
    --- End diff --
    
    this list does not contain the "expected" values.


> Make the GenericWriteAheadSink rescalable.
> ------------------------------------------
>
>                 Key: FLINK-5020
>                 URL: https://issues.apache.org/jira/browse/FLINK-5020
>             Project: Flink
>          Issue Type: Improvement
>          Components: Cassandra Connector
>            Reporter: Kostas Kloudas
>            Assignee: Kostas Kloudas
>             Fix For: 1.2.0
>
>
> This targets integrating the GenericWriteAheadSink with the new rescalable state abstractions
so that the parallelism of the operator can change arbitrarily without jeopardizing the guarantees
offered by it.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message