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:16:59 GMT

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

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_r86754225
  
    --- Diff: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/WriteAheadSinkTestBase.java
---
    @@ -158,17 +160,151 @@ public void testDataDiscardingUponRestore() throws Exception {
     		testHarness = new OneInputStreamOperatorTestHarness<>(sink);
     
     		testHarness.setup();
    -		testHarness.restore(latestSnapshot);
    +		testHarness.initializeState(latestSnapshot);
     		testHarness.open();
     
     		for (int x = 0; x < 20; x++) {
     			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter,
2)));
     			elementCounter++;
     		}
     
    -		testHarness.snapshotLegacy(snapshotCount++, 0);
    +		testHarness.snapshot(snapshotCount++, 0);
     		testHarness.notifyOfCompletedCheckpoint(snapshotCount - 1);
     
    -		verifyResultsDataDiscardingUponRestore(testHarness, sink);
    +		verifyResultsDataDiscardingUponRestore(sink);
    +	}
    +
    +	@Test
    +	public void testScalingDown() throws Exception {
    +		S sink1 = createSink();
    +		OneInputStreamOperatorTestHarness<IN, IN> testHarness1 =
    +			new OneInputStreamOperatorTestHarness<>(sink1, 10, 2, 0);
    +		testHarness1.open();
    +
    +		S sink2 = createSink();
    +		OneInputStreamOperatorTestHarness<IN, IN> testHarness2 =
    +			new OneInputStreamOperatorTestHarness<>(sink2, 10, 2, 1);
    +		testHarness2.open();
    +
    +		int elementCounter = 1;
    +		int snapshotCount = 0;
    +
    +		for (int x = 0; x < 10; x++) {
    +			testHarness1.processElement(new StreamRecord<>(generateValue(elementCounter,
0)));
    +			elementCounter++;
    +		}
    +
    +		for (int x = 0; x < 11; x++) {
    +			testHarness2.processElement(new StreamRecord<>(generateValue(elementCounter,
0)));
    +			elementCounter++;
    +		}
    +
    +		// snapshot at checkpoint 0 for testHarness1 and testHarness 2
    +		OperatorStateHandles snapshot1 = testHarness1.snapshot(snapshotCount, 0);
    +		OperatorStateHandles snapshot2 = testHarness2.snapshot(snapshotCount, 0);
    +
    +		// merge the two partial states
    +		OperatorStateHandles mergedSnapshot = AbstractStreamOperatorTestHarness
    +			.repackageState(snapshot1, snapshot2);
    +
    +		testHarness1.close();
    +		testHarness2.close();
    +
    +		// and create a third instance that operates alone but
    +		// has the merged state of the previous 2 instances
    +
    +		S sink3 = createSink();
    +		OneInputStreamOperatorTestHarness<IN, IN> mergedTestHarness =
    +			new OneInputStreamOperatorTestHarness<>(sink3, 10, 1, 0);
    +
    +		mergedTestHarness.setup();
    +		mergedTestHarness.initializeState(mergedSnapshot);
    +		mergedTestHarness.open();
    +
    +		for (int x = 0; x < 12; x++) {
    +			mergedTestHarness.processElement(new StreamRecord<>(generateValue(elementCounter,
0)));
    +			elementCounter++;
    +		}
    +
    +		snapshotCount++;
    +		mergedTestHarness.snapshot(snapshotCount, 1);
    +		mergedTestHarness.notifyOfCompletedCheckpoint(snapshotCount);
    +
    +		verifyResultsWhenScalingDown(sink3);
    +		mergedTestHarness.close();
    +	}
    +
    +	@Test
    +	public void testScalingUp() throws Exception {
    +
    +		S sink1 = createSink();
    +		OneInputStreamOperatorTestHarness<IN, IN> testHarness1 =
    +			new OneInputStreamOperatorTestHarness<>(sink1, 10, 1, 0);
    +
    +		int elementCounter = 1;
    +		int snapshotCount = 0;
    +
    +		testHarness1.open();
    +
    +		// put two more checkpoints as pending
    +
    +		for (int x = 0; x < 10; x++) {
    +			testHarness1.processElement(new StreamRecord<>(generateValue(elementCounter,
0)));
    +			elementCounter++;
    +		}
    +		testHarness1.snapshot(++snapshotCount, 0);
    +
    +		for (int x = 0; x < 11; x++) {
    +			testHarness1.processElement(new StreamRecord<>(generateValue(elementCounter,
0)));
    +			elementCounter++;
    +		}
    +
    +		// this will be the state that will be split between the two new operators
    +		OperatorStateHandles snapshot = testHarness1.snapshot(++snapshotCount, 0);
    +
    +		testHarness1.close();
    +
    +		// verify no elements are in the sink
    +		verifyResultsWhenScalingUp(sink1, 0, -1);
    --- End diff --
    
    this isn't necessary


> 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