[FLINK-5101] Refactor CassandraConnectorITCase #2866

zentol · 2016-11-25T10:57:04Z

This PR refactors the CassandraConnectorITCase to be a bit more stable and easier to debug.

The following changes were made:

we no longer run actual flink jobs; all tests directly interact with the sink to save resources
every test uses a different table, preventing race conditions related to truncating the table
the at-least-once sinks were modified to track pending updates
=> the pojo sink was modified to use a method that returns an actually useful Future
=> since the sink waits in close() for pending updates it can no longer occur that a test checks a condition prematurely, improving stability
the initial connection is established across a time-span of 30 seconds, increasing the chance that cassandra has started before the tests are run

StephanEwen · 2017-01-23T12:37:05Z

...ssandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBase.java

+				try {
+					updatesPending.wait();
+				} catch (InterruptedException e) {
+				}


Good practice is to set the interruption flag back: Thread.currentThread().interrupt();

Probably better here even: Throw an exception, because closing is incomplete when interrupted (and cannot guarantee correctness)

StephanEwen · 2017-01-23T12:41:15Z

...ssandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBase.java

@@ -94,5 +122,9 @@ public void close() {
 		} catch (Exception e) {
 			LOG.error("Error while closing cluster.", e);
 		}
+		Throwable e = exception.get();
+		if (e != null) {
+			LOG.error("Error while sending value.", e);


This needs to be rethrown, otherwise close() may complete without an exception, but the result may be incomplete.

StephanEwen · 2017-01-23T12:46:00Z

.../src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java

-		try {
-			Thread.sleep(1000 * 10);
-		} catch (InterruptedException e) { //give cassandra a few seconds to start up
+		// give cassandra a few seconds to start up


Can we skip this and simply rely on the loop below that tries to connect?

StephanEwen · 2017-01-23T12:46:29Z

.../src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java

+					throw e;
+				}
+				try {
+					Thread.sleep(2000);


How about reducing this to, say 500ms, to make it react a bit sooner once the connection can be established.

StephanEwen · 2017-01-23T12:47:05Z

.../src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java

-		cluster = builder.getCluster();
-		session = cluster.connect();
+		// start establishing a connection within 30 seconds
+		start = System.currentTimeMillis();


I would suggest to use System.nanoTime(), because System.curremtTimeMillis is not stable. Seems especially unstable on Travis.

StephanEwen · 2017-01-23T12:49:49Z

.../src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java


-		ResultSet rs = session.execute(SELECT_DATA_QUERY);
+		synchronized (sink.updatesPending) {


Is this needed when close() already waits for pending requests?

StephanEwen · 2017-01-23T12:50:05Z

.../src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java

+
+		sink.close();
+
+		synchronized (sink.updatesPending) {


Same as above

StephanEwen · 2017-01-23T12:50:50Z

Good set of changes. Some comments for small adjustments, otherwise +1 for the approch

zentol · 2017-01-26T10:54:00Z

@StephanEwen I've addressed your comments and Travis is passing :)

StephanEwen · 2017-01-26T15:34:18Z

.../src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java

-		} catch (InterruptedException e) { //give cassandra a few seconds to start up
+		// start establishing a connection within 30 seconds
+		long start = System.nanoTime();
+		long deadline = start + 1000 * 30;


This needs to be long deadline = start + 30_000_000_000 (because its nanos)

StephanEwen · 2017-01-26T15:34:46Z

.../src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java

+				session = cluster.connect();
+				break;
+			} catch (Exception e) {
+				if (System.currentTimeMillis() > deadline) {


This should also be System.nanoTime()

StephanEwen · 2017-01-26T16:07:28Z

...ssandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBase.java

@@ -40,26 +42,42 @@
 	protected transient Cluster cluster;
 	protected transient Session session;

-	protected transient Throwable exception = null;
+	protected transient AtomicReference<Throwable> exception;


Does exception have to be an AtomicReference, or does a volatile variable suffice?

it can be volatile.

StephanEwen · 2017-01-26T16:08:40Z

Looks much better, one mixup is still in the tests.
The comment about the Atomic Reference is a minor improvement comment.

zentol · 2017-02-22T16:21:43Z

@StephanEwen I've addressed your comments and rebased the branch.

zentol · 2017-05-10T10:01:58Z

Will start merging this; it's blocking #3810.

This closes apache#2866.

StephanEwen requested changes Jan 23, 2017

View reviewed changes

zentol force-pushed the 4177_cass_test branch from 4c74937 to 350a75c Compare January 23, 2017 14:13

StephanEwen requested changes Jan 26, 2017

View reviewed changes

zentol added 6 commits February 6, 2017 12:36

[FLINK-5101] Track pending records in CassandraSinkBase

2878d58

[FLINK-5101] Refactor CassandraConnectorITCase

985b921

address comments

5bf9e1d

+

936f1cc

fix conversion to nano-time

3c5c7c4

replace AtomicReference<Throwable> with volative

9a75b6e

zentol force-pushed the 4177_cass_test branch from 364f96e to 9a75b6e Compare February 6, 2017 11:38

zentol added a commit to zentol/flink that referenced this pull request May 11, 2017

[FLINK-5101] Refactor CassandraConnectorITCase

2ee1a4b

This closes apache#2866.

zentol added a commit to zentol/flink that referenced this pull request May 12, 2017

[FLINK-5101] Refactor CassandraConnectorITCase

84ced25

This closes apache#2866.

zentol added a commit to zentol/flink that referenced this pull request May 12, 2017

[FLINK-5101] Refactor CassandraConnectorITCase

35042f7

This closes apache#2866.

asfgit closed this in 0be04b4 May 12, 2017

fanyon pushed a commit to fanyon/flink that referenced this pull request May 15, 2017

[FLINK-5101] Refactor CassandraConnectorITCase

1261451

This closes apache#2866.

zentol deleted the 4177_cass_test branch November 6, 2017 14:47

rmetzger added the component=Connectors/Cassandra label Mar 14, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-5101] Refactor CassandraConnectorITCase #2866

[FLINK-5101] Refactor CassandraConnectorITCase #2866

zentol commented Nov 25, 2016

StephanEwen Jan 23, 2017

StephanEwen Jan 23, 2017

StephanEwen Jan 23, 2017

StephanEwen Jan 23, 2017

StephanEwen Jan 23, 2017

StephanEwen Jan 23, 2017

StephanEwen Jan 23, 2017

StephanEwen Jan 23, 2017

StephanEwen commented Jan 23, 2017

zentol commented Jan 26, 2017

StephanEwen Jan 26, 2017

StephanEwen Jan 26, 2017

StephanEwen Jan 26, 2017

zentol Feb 6, 2017

StephanEwen commented Jan 26, 2017

zentol commented Feb 22, 2017

zentol commented May 10, 2017


		ResultSet rs = session.execute(SELECT_DATA_QUERY);
		synchronized (sink.updatesPending) {

[FLINK-5101] Refactor CassandraConnectorITCase #2866

[FLINK-5101] Refactor CassandraConnectorITCase #2866

Conversation

zentol commented Nov 25, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

StephanEwen commented Jan 23, 2017

zentol commented Jan 26, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

StephanEwen commented Jan 26, 2017

zentol commented Feb 22, 2017

zentol commented May 10, 2017