You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2016/07/21 13:21:44 UTC
flink git commit: [FLINK-4238] Only allow/require query for Tuple
Stream in CassandraSink
Repository: flink
Updated Branches:
refs/heads/master f81cda38b -> d3bc5563b
[FLINK-4238] Only allow/require query for Tuple Stream in CassandraSink
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d3bc5563
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d3bc5563
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d3bc5563
Branch: refs/heads/master
Commit: d3bc5563b26c2cf10052dda742790ddf73adc00e
Parents: f81cda3
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Wed Jul 20 18:10:11 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Thu Jul 21 15:21:13 2016 +0200
----------------------------------------------------------------------
.../connectors/cassandra/CassandraSink.java | 19 ++++++++++++++++---
1 file changed, 16 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d3bc5563/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java
index c01f392..62b0c81 100644
--- a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java
@@ -271,9 +271,6 @@ public class CassandraSink<IN> {
public abstract CassandraSink<IN> build() throws Exception;
protected void sanityCheck() {
- if (query == null || query.length() == 0) {
- throw new IllegalArgumentException("Query must not be null or empty.");
- }
if (builder == null) {
throw new IllegalArgumentException("Cassandra host information must be supplied using either setHost() or setClusterBuilder().");
}
@@ -286,6 +283,14 @@ public class CassandraSink<IN> {
}
@Override
+ protected void sanityCheck() {
+ super.sanityCheck();
+ if (query == null || query.length() == 0) {
+ throw new IllegalArgumentException("Query must not be null or empty.");
+ }
+ }
+
+ @Override
public CassandraSink<IN> build() throws Exception {
sanityCheck();
if (isWriteAheadLogEnabled) {
@@ -304,6 +309,14 @@ public class CassandraSink<IN> {
}
@Override
+ protected void sanityCheck() {
+ super.sanityCheck();
+ if (query != null) {
+ throw new IllegalArgumentException("Specifying a query is not allowed when using a Pojo-Stream as input.");
+ }
+ }
+
+ @Override
public CassandraSink<IN> build() throws Exception {
sanityCheck();
if (isWriteAheadLogEnabled) {