You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2019/07/03 22:34:55 UTC
[flink] branch master updated (2b1a125 -> 6624562)
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 2b1a125 [hotfix][table] remove @PublicEvolving annotation from AbstractCatalog as it's not supported to be public
new 085a7ba [FLINK-12974] [build] Bump checkstyle to 8.14
new b5f4f32 [hotfix] [docs] Update IDE setup instructions for latest IntelliJ IDEA
new 3854552 [FLINK-12840] [core] Fix network utils to work with ipv6 correctly
new 5c9caa7 [hotfix] [examples] Distinguish 'netcat' arguments for Linux and Windows in the SocketWIndowWordcount JavaDocs
new 7a8333b [FLINK-13017] [docs] Do not mount local $HOME into docs docker environment
new a125e2a [FLINK-12820] [Connectors / Cassandra] Support ignoring writing nulls for tuple types
new 6624562 [FLINK-13043] [Library / CEP] Fix the bug of parsing Dewey number from string
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/docker/run.sh | 3 ---
docs/flinkDev/ide_setup.md | 25 +++++++++---------
docs/flinkDev/ide_setup.zh.md | 23 +++++++++--------
.../cassandra/AbstractCassandraTupleSink.java | 17 +++++++++++-
.../connectors/cassandra/CassandraSink.java | 13 ++++++++++
.../cassandra/CassandraSinkBaseConfig.java | 27 +++++++++++++++++--
.../cassandra/CassandraConnectorITCase.java | 30 ++++++++++++++++++++++
.../main/java/org/apache/flink/util/NetUtils.java | 6 +++++
.../java/org/apache/flink/util/NetUtilsTest.java | 6 +++++
.../examples/socket/SocketWindowWordCount.java | 2 +-
.../examples/socket/SocketWindowWordCount.scala | 2 +-
.../java/org/apache/flink/cep/nfa/DeweyNumber.java | 6 +++--
.../org/apache/flink/cep/nfa/DeweyNumberTest.java | 5 ++++
.../org/apache/flink/runtime/akka/AkkaUtils.scala | 6 ++---
.../apache/flink/runtime/akka/AkkaUtilsTest.scala | 11 ++++++++
pom.xml | 2 +-
16 files changed, 147 insertions(+), 37 deletions(-)
mode change 100644 => 100755 flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
[flink] 07/07: [FLINK-13043] [Library / CEP] Fix the bug of parsing
Dewey number from string
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6624562982c9d57bebba8cb4b574b8ed28640a0d
Author: liyafan82 <fa...@foxmail.com>
AuthorDate: Mon Jul 1 18:59:40 2019 +0800
[FLINK-13043] [Library / CEP] Fix the bug of parsing Dewey number from string
This closes #8936
---
.../src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java | 6 ++++--
.../src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java | 5 +++++
2 files changed, 9 insertions(+), 2 deletions(-)
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java
index 6141bf2..de41ef1 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java
@@ -175,9 +175,9 @@ public class DeweyNumber implements Serializable {
public static DeweyNumber fromString(final String deweyNumberString) {
String[] splits = deweyNumberString.split("\\.");
- if (splits.length == 0) {
+ if (splits.length == 1) {
return new DeweyNumber(Integer.parseInt(deweyNumberString));
- } else {
+ } else if (splits.length > 0) {
int[] deweyNumber = new int[splits.length];
for (int i = 0; i < splits.length; i++) {
@@ -185,6 +185,8 @@ public class DeweyNumber implements Serializable {
}
return new DeweyNumber(deweyNumber);
+ } else {
+ throw new IllegalArgumentException("Failed to parse " + deweyNumberString + " as a Dewey number");
}
}
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java
index e28e77d..bf07294 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/DeweyNumberTest.java
@@ -55,4 +55,9 @@ public class DeweyNumberTest extends TestLogger {
assertFalse(startAddStage.isCompatibleWith(increaseAddStage));
assertFalse(startAddStageIncreased.isCompatibleWith(startAddStageIncreasedAddStage));
}
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testZeroSplitsDeweyNumber() {
+ DeweyNumber.fromString(".");
+ }
}
[flink] 02/07: [hotfix] [docs] Update IDE setup instructions for
latest IntelliJ IDEA
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b5f4f327f6537f1ec796d35c7ee1f2a65dab3565
Author: Hugo Da Cruz Louro <hm...@apache.org>
AuthorDate: Wed Jun 26 23:46:31 2019 -0700
[hotfix] [docs] Update IDE setup instructions for latest IntelliJ IDEA
This closes #8908
---
docs/flinkDev/ide_setup.md | 23 ++++++++++++-----------
docs/flinkDev/ide_setup.zh.md | 23 ++++++++++++-----------
2 files changed, 24 insertions(+), 22 deletions(-)
diff --git a/docs/flinkDev/ide_setup.md b/docs/flinkDev/ide_setup.md
index c6b0cd2..c27e479 100644
--- a/docs/flinkDev/ide_setup.md
+++ b/docs/flinkDev/ide_setup.md
@@ -49,7 +49,7 @@ git clone https://github.com/apache/flink.git
A brief guide on how to set up IntelliJ IDEA IDE for development of the Flink core.
As Eclipse is known to have issues with mixed Scala and Java projects, more and more contributors are migrating to IntelliJ IDEA.
-The following documentation describes the steps to setup IntelliJ IDEA 2016.2.5
+The following documentation describes the steps to setup IntelliJ IDEA 2019.1.3
([https://www.jetbrains.com/idea/download/](https://www.jetbrains.com/idea/download/))
with the Flink sources.
@@ -66,19 +66,20 @@ to enable support for Scala projects and files:
### Importing Flink
-1. Start IntelliJ IDEA and choose "Import Project"
-2. Select the root folder of the Flink repository
+1. Start IntelliJ IDEA and choose "New -> Project from Existing Sources"
+2. Select the root folder of the cloned Flink repository
3. Choose "Import project from external model" and select "Maven"
-4. Leave the default options and click on "Next" until you hit the SDK section.
-5. If there is no SDK, create a one with the "+" sign top left,
- then click "JDK", select your JDK home directory and click "OK".
- Otherwise simply select your SDK.
-6. Continue by clicking "Next" again and finish the import.
+4. Leave the default options and successively click "Next" until you reach the SDK section.
+5. If there is no SDK listed, create one using the "+" sign on the top left.
+ Select "JDK", choose the JDK home directory and click "OK".
+ Select the most suiting JDK version. NOTE: A good rule of thumb is to select
+ the JDK version matching the active Maven profile.
+6. Continue by clicking "Next" until finishing the import.
7. Right-click on the imported Flink project -> Maven -> Generate Sources and Update Folders.
Note that this will install Flink libraries in your local Maven repository,
- i.e. "/home/*-your-user-*/.m2/repository/org/apache/flink/".
- Alternatively, `mvn clean package -DskipTests` also creates the necessary
- files for the IDE to work with but without installing libraries.
+ located by default at "/home/$USER/.m2/repository/org/apache/flink/".
+ Alternatively, `mvn clean package -DskipTests` also creates the files necessary
+ for the IDE to work but without installing the libraries.
8. Build the Project (Build -> Make Project)
### Checkstyle For Java
diff --git a/docs/flinkDev/ide_setup.zh.md b/docs/flinkDev/ide_setup.zh.md
index a893d66..1177c6a 100644
--- a/docs/flinkDev/ide_setup.zh.md
+++ b/docs/flinkDev/ide_setup.zh.md
@@ -49,7 +49,7 @@ git clone https://github.com/apache/flink.git
A brief guide on how to set up IntelliJ IDEA IDE for development of the Flink core.
As Eclipse is known to have issues with mixed Scala and Java projects, more and more contributors are migrating to IntelliJ IDEA.
-The following documentation describes the steps to setup IntelliJ IDEA 2016.2.5
+The following documentation describes the steps to setup IntelliJ IDEA 2019.1.3
([https://www.jetbrains.com/idea/download/](https://www.jetbrains.com/idea/download/))
with the Flink sources.
@@ -66,19 +66,20 @@ to enable support for Scala projects and files:
### Importing Flink
-1. Start IntelliJ IDEA and choose "Import Project"
-2. Select the root folder of the Flink repository
+1. Start IntelliJ IDEA and choose "New -> Project from Existing Sources"
+2. Select the root folder of the cloned Flink repository
3. Choose "Import project from external model" and select "Maven"
-4. Leave the default options and click on "Next" until you hit the SDK section.
-5. If there is no SDK, create a one with the "+" sign top left,
- then click "JDK", select your JDK home directory and click "OK".
- Otherwise simply select your SDK.
-6. Continue by clicking "Next" again and finish the import.
+4. Leave the default options and successively click "Next" until you reach the SDK section.
+5. If there is no SDK listed, create one using the "+" sign on the top left.
+ Select "JDK", choose the JDK home directory and click "OK".
+ Select the most suiting JDK version. NOTE: A good rule of thumb is to select
+ the JDK version matching the active Maven profile.
+6. Continue by clicking "Next" until finishing the import.
7. Right-click on the imported Flink project -> Maven -> Generate Sources and Update Folders.
Note that this will install Flink libraries in your local Maven repository,
- i.e. "/home/*-your-user-*/.m2/repository/org/apache/flink/".
- Alternatively, `mvn clean package -DskipTests` also creates the necessary
- files for the IDE to work with but without installing libraries.
+ located by default at "/home/$USER/.m2/repository/org/apache/flink/".
+ Alternatively, `mvn clean package -DskipTests` also creates the files necessary
+ for the IDE to work but without installing the libraries.
8. Build the Project (Build -> Make Project)
### Checkstyle For Java
[flink] 05/07: [FLINK-13017] [docs] Do not mount local $HOME into
docs docker environment
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7a8333b18503cc2414bb405dced91d4b83260a6d
Author: Nico Kruber <ni...@ververica.com>
AuthorDate: Thu Jun 27 14:39:33 2019 +0200
[FLINK-13017] [docs] Do not mount local $HOME into docs docker environment
This closes #8917
---
docs/docker/run.sh | 3 ---
1 file changed, 3 deletions(-)
diff --git a/docs/docker/run.sh b/docs/docker/run.sh
index 5598c0a..bd02178 100755
--- a/docs/docker/run.sh
+++ b/docs/docker/run.sh
@@ -31,12 +31,10 @@ if [ "$(uname -s)" == "Linux" ]; then
USER_NAME=${SUDO_USER:=$USER}
USER_ID=$(id -u "${USER_NAME}")
GROUP_ID=$(id -g "${USER_NAME}")
- LOCAL_HOME="/home/${USER_NAME}"
else # boot2docker uid and gid
USER_NAME=$USER
USER_ID=1000
GROUP_ID=50
- LOCAL_HOME="/Users/${USER_NAME}"
fi
docker build -t "${IMAGE_NAME}-${USER_NAME}" - <<UserSpecificDocker
@@ -67,7 +65,6 @@ docker run -i -t \
-w ${FLINK_DOC_ROOT} \
-u "${USER}" \
-v "${FLINK_DOC_ROOT}:${FLINK_DOC_ROOT}" \
- -v "${LOCAL_HOME}:/home/${USER_NAME}" \
-p 4000:4000 \
${IMAGE_NAME}-${USER_NAME} \
bash -c "${CMD}"
[flink] 03/07: [FLINK-12840] [core] Fix network utils to work with
ipv6 correctly
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3854552ceefd2b2b9c0e2a9b6152a7fcb69153fe
Author: potseluev <po...@yandex-team.ru>
AuthorDate: Fri Jun 14 01:04:44 2019 +0300
[FLINK-12840] [core] Fix network utils to work with ipv6 correctly
- Fixes problems around akka configuration parsing with some IPv6 literals
- Fixes an issue with address parsing and validation with some Ipv6 literals
This closes #8734
---
flink-core/src/main/java/org/apache/flink/util/NetUtils.java | 6 ++++++
.../src/test/java/org/apache/flink/util/NetUtilsTest.java | 6 ++++++
.../main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala | 6 +++---
.../scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala | 11 +++++++++++
4 files changed, 26 insertions(+), 3 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/NetUtils.java b/flink-core/src/main/java/org/apache/flink/util/NetUtils.java
index 447e8a9..1a4bb7a 100644
--- a/flink-core/src/main/java/org/apache/flink/util/NetUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/util/NetUtils.java
@@ -134,6 +134,12 @@ public class NetUtils {
host = InetAddress.getLoopbackAddress().getHostAddress();
} else {
host = host.trim().toLowerCase();
+ if (host.startsWith("[") && host.endsWith("]")) {
+ String address = host.substring(1, host.length() - 1);
+ if (IPAddressUtil.isIPv6LiteralAddress(address)) {
+ host = address;
+ }
+ }
}
// normalize and valid address
diff --git a/flink-core/src/test/java/org/apache/flink/util/NetUtilsTest.java b/flink-core/src/test/java/org/apache/flink/util/NetUtilsTest.java
index 04faa67..ffd9e5e 100644
--- a/flink-core/src/test/java/org/apache/flink/util/NetUtilsTest.java
+++ b/flink-core/src/test/java/org/apache/flink/util/NetUtilsTest.java
@@ -193,6 +193,12 @@ public class NetUtilsTest {
Assert.assertEquals("[2001:db8:85a3::8a2e:370:7334]:" + port, NetUtils.unresolvedHostAndPortToNormalizedString(host, port));
}
{
+ // [IPv6]
+ String host = "[2001:0db8:85a3:0000:0000:8a2e:0370:7334]";
+ int port = 42;
+ Assert.assertEquals("[2001:db8:85a3::8a2e:370:7334]:" + port, NetUtils.unresolvedHostAndPortToNormalizedString(host, port));
+ }
+ {
// Hostnames
String host = "somerandomhostname";
int port = 99;
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
old mode 100644
new mode 100755
index 14563e9..e9b11f7
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
@@ -547,7 +547,7 @@ object AkkaUtils {
} else {
// if bindAddress is null or empty, then leave bindAddress unspecified. Akka will pick
// InetAddress.getLocalHost.getHostAddress
- "\"\""
+ ""
}
val hostnameConfigString =
@@ -556,8 +556,8 @@ object AkkaUtils {
| remote {
| netty {
| tcp {
- | hostname = $effectiveHostname
- | bind-hostname = $bindAddress
+ | hostname = "$effectiveHostname"
+ | bind-hostname = "$bindAddress"
| }
| }
| }
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala
index c051f7a..8c970a0 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaUtilsTest.scala
@@ -209,4 +209,15 @@ class AkkaUtilsTest
akkaConfig.getInt("akka.actor.default-dispatcher.thread-pool-executor.core-pool-size-max")
.should(equal(maxThreads))
}
+
+ test("getAkkaConfig should work with ipv6 addresses") {
+ val ipv6AddressString = "2001:db8:10:11:12:ff00:42:8329"
+ val configuration = new Configuration()
+ val port = 1234
+
+ val akkaConfig = AkkaUtils.getAkkaConfig(configuration, ipv6AddressString, port)
+
+ akkaConfig.getString("akka.remote.netty.tcp.hostname") should
+ equal(NetUtils.unresolvedHostToNormalizedString(ipv6AddressString))
+ }
}
[flink] 01/07: [FLINK-12974] [build] Bump checkstyle to 8.14
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 085a7ba64bc1325717bd5f882d308eb6b3dd9ee5
Author: ifndef-SleePy <mm...@gmail.com>
AuthorDate: Tue Jun 25 17:06:59 2019 +0800
[FLINK-12974] [build] Bump checkstyle to 8.14
This closes #8870
---
docs/flinkDev/ide_setup.md | 2 +-
pom.xml | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/flinkDev/ide_setup.md b/docs/flinkDev/ide_setup.md
index 3dfc905..c6b0cd2 100644
--- a/docs/flinkDev/ide_setup.md
+++ b/docs/flinkDev/ide_setup.md
@@ -87,7 +87,7 @@ IntelliJ supports checkstyle within the IDE using the Checkstyle-IDEA plugin.
1. Install the "Checkstyle-IDEA" plugin from the IntelliJ plugin repository.
2. Configure the plugin by going to Settings -> Other Settings -> Checkstyle.
3. Set the "Scan Scope" to "Only Java sources (including tests)".
-4. Select _8.12_ in the "Checkstyle Version" dropdown and click apply. **This step is important,
+4. Select _8.14_ in the "Checkstyle Version" dropdown and click apply. **This step is important,
don't skip it!**
5. In the "Configuration File" pane, add a new configuration using the plus icon:
1. Set the "Description" to "Flink".
diff --git a/pom.xml b/pom.xml
index 2b3fa48..0357c7a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1608,7 +1608,7 @@ under the License.
<groupId>com.puppycrawl.tools</groupId>
<artifactId>checkstyle</artifactId>
<!-- Note: match version with docs/flinkDev/ide_setup.md -->
- <version>8.12</version>
+ <version>8.14</version>
</dependency>
</dependencies>
<executions>
[flink] 06/07: [FLINK-12820] [Connectors / Cassandra] Support
ignoring writing nulls for tuple types
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a125e2a11444847920911927955be2d119716838
Author: ozan <oz...@gmail.com>
AuthorDate: Wed Jun 12 21:35:54 2019 +0900
[FLINK-12820] [Connectors / Cassandra] Support ignoring writing nulls for tuple types
This closes #8714
---
.../cassandra/AbstractCassandraTupleSink.java | 17 +++++++++++-
.../connectors/cassandra/CassandraSink.java | 13 ++++++++++
.../cassandra/CassandraSinkBaseConfig.java | 27 +++++++++++++++++--
.../cassandra/CassandraConnectorITCase.java | 30 ++++++++++++++++++++++
4 files changed, 84 insertions(+), 3 deletions(-)
diff --git a/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/AbstractCassandraTupleSink.java b/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/AbstractCassandraTupleSink.java
index 5e1fcca..3cf8aa4 100644
--- a/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/AbstractCassandraTupleSink.java
+++ b/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/AbstractCassandraTupleSink.java
@@ -19,6 +19,7 @@ package org.apache.flink.streaming.connectors.cassandra;
import org.apache.flink.configuration.Configuration;
+import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.google.common.util.concurrent.ListenableFuture;
@@ -31,6 +32,7 @@ import com.google.common.util.concurrent.ListenableFuture;
public abstract class AbstractCassandraTupleSink<IN> extends CassandraSinkBase<IN, ResultSet> {
private final String insertQuery;
private transient PreparedStatement ps;
+ private final boolean ignoreNullFields;
public AbstractCassandraTupleSink(
String insertQuery,
@@ -39,6 +41,7 @@ public abstract class AbstractCassandraTupleSink<IN> extends CassandraSinkBase<I
CassandraFailureHandler failureHandler) {
super(builder, config, failureHandler);
this.insertQuery = insertQuery;
+ this.ignoreNullFields = config.getIgnoreNullFields();
}
@Override
@@ -50,7 +53,19 @@ public abstract class AbstractCassandraTupleSink<IN> extends CassandraSinkBase<I
@Override
public ListenableFuture<ResultSet> send(IN value) {
Object[] fields = extract(value);
- return session.executeAsync(ps.bind(fields));
+ return session.executeAsync(bind(fields));
+ }
+
+ private BoundStatement bind(Object[] fields) {
+ BoundStatement bs = ps.bind(fields);
+ if (ignoreNullFields) {
+ for (int i = 0; i < fields.length; i++) {
+ if (fields[i] == null) {
+ bs.unset(i);
+ }
+ }
+ }
+ return bs;
}
protected abstract Object[] extract(IN record);
diff --git a/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java b/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java
index 128e5e0..2a8ebff 100644
--- a/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java
+++ b/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSink.java
@@ -399,6 +399,19 @@ public class CassandraSink<IN> {
}
/**
+ * Enables ignoring null values, treats null values as unset and avoids writing null fields
+ * and creating tombstones.
+ *
+ * <p>This call has no effect if {@link CassandraSinkBuilder#enableWriteAheadLog()} is called.
+ *
+ * @return this builder
+ */
+ public CassandraSinkBuilder<IN> enableIgnoreNullFields() {
+ this.configBuilder.setIgnoreNullFields(true);
+ return this;
+ }
+
+ /**
* Finalizes the configuration of this sink.
*
* @return finalized sink
diff --git a/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBaseConfig.java b/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBaseConfig.java
index cb8d904..d48f973 100644
--- a/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBaseConfig.java
+++ b/flink-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBaseConfig.java
@@ -38,6 +38,12 @@ public final class CassandraSinkBaseConfig implements Serializable {
*/
public static final Duration DEFAULT_MAX_CONCURRENT_REQUESTS_TIMEOUT = Duration.ofMillis(Long.MAX_VALUE);
+ /**
+ * The default option to ignore null fields on insertion. By default, {@code false}.
+ */
+ public static final boolean DEFAULT_IGNORE_NULL_FIELDS = false;
+
+
// ------------------------- Configuration Fields -------------------------
/** Maximum number of concurrent requests allowed. */
@@ -46,9 +52,13 @@ public final class CassandraSinkBaseConfig implements Serializable {
/** Timeout duration when acquiring a permit to execute. */
private final Duration maxConcurrentRequestsTimeout;
+ /** Whether to ignore null fields on insert. */
+ private final boolean ignoreNullFields;
+
private CassandraSinkBaseConfig(
int maxConcurrentRequests,
- Duration maxConcurrentRequestsTimeout) {
+ Duration maxConcurrentRequestsTimeout,
+ boolean ignoreNullFields) {
Preconditions.checkArgument(maxConcurrentRequests > 0,
"Max concurrent requests is expected to be positive");
Preconditions.checkNotNull(maxConcurrentRequestsTimeout,
@@ -57,6 +67,7 @@ public final class CassandraSinkBaseConfig implements Serializable {
"Max concurrent requests timeout is expected to be positive");
this.maxConcurrentRequests = maxConcurrentRequests;
this.maxConcurrentRequestsTimeout = maxConcurrentRequestsTimeout;
+ this.ignoreNullFields = ignoreNullFields;
}
public int getMaxConcurrentRequests() {
@@ -67,11 +78,16 @@ public final class CassandraSinkBaseConfig implements Serializable {
return maxConcurrentRequestsTimeout;
}
+ public boolean getIgnoreNullFields() {
+ return ignoreNullFields;
+ }
+
@Override
public String toString() {
return "CassandraSinkBaseConfig{" +
"maxConcurrentRequests=" + maxConcurrentRequests +
", maxConcurrentRequestsTimeout=" + maxConcurrentRequestsTimeout +
+ ", ignoreNullFields=" + ignoreNullFields +
'}';
}
@@ -85,6 +101,7 @@ public final class CassandraSinkBaseConfig implements Serializable {
public static class Builder {
private int maxConcurrentRequests = DEFAULT_MAX_CONCURRENT_REQUESTS;
private Duration maxConcurrentRequestsTimeout = DEFAULT_MAX_CONCURRENT_REQUESTS_TIMEOUT;
+ private boolean ignoreNullFields = DEFAULT_IGNORE_NULL_FIELDS;
Builder() { }
@@ -98,10 +115,16 @@ public final class CassandraSinkBaseConfig implements Serializable {
return this;
}
+ public Builder setIgnoreNullFields(boolean ignoreNullFields) {
+ this.ignoreNullFields = ignoreNullFields;
+ return this;
+ }
+
public CassandraSinkBaseConfig build() {
return new CassandraSinkBaseConfig(
maxConcurrentRequests,
- maxConcurrentRequestsTimeout);
+ maxConcurrentRequestsTimeout,
+ ignoreNullFields);
}
}
}
diff --git a/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java b/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
index b3d8f65..0119c16 100644
--- a/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
+++ b/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
@@ -637,4 +637,34 @@ public class CassandraConnectorITCase extends WriteAheadSinkTestBase<Tuple3<Stri
}
Assert.assertEquals(0, scalaTupleCollection.size());
}
+
+ @Test
+ public void testCassandraScalaTuplePartialColumnUpdate() throws Exception {
+ CassandraSinkBaseConfig config = CassandraSinkBaseConfig.newBuilder().setIgnoreNullFields(true).build();
+ CassandraScalaProductSink<scala.Tuple3<String, Integer, Integer>> sink = new CassandraScalaProductSink<>(injectTableName(INSERT_DATA_QUERY), builder, config);
+
+ String id = UUID.randomUUID().toString();
+ Integer counter = 1;
+ Integer batchId = 0;
+
+ // Send partial records across multiple request
+ scala.Tuple3<String, Integer, Integer> scalaTupleRecordFirst = new scala.Tuple3<>(id, counter, null);
+ scala.Tuple3<String, Integer, Integer> scalaTupleRecordSecond = new scala.Tuple3<>(id, null, batchId);
+
+ try {
+ sink.open(new Configuration());
+ sink.invoke(scalaTupleRecordFirst, SinkContextUtil.forTimestamp(0));
+ sink.invoke(scalaTupleRecordSecond, SinkContextUtil.forTimestamp(0));
+ } finally {
+ sink.close();
+ }
+
+ ResultSet rs = session.execute(injectTableName(SELECT_DATA_QUERY));
+ List<com.datastax.driver.core.Row> rows = rs.all();
+ Assert.assertEquals(1, rows.size());
+ // Since nulls are ignored, we should be reading one complete record
+ for (com.datastax.driver.core.Row row : rows) {
+ Assert.assertEquals(new scala.Tuple3<>(id, counter, batchId), new scala.Tuple3<>(row.getString("id"), row.getInt("counter"), row.getInt("batch_id")));
+ }
+ }
}
[flink] 04/07: [hotfix] [examples] Distinguish 'netcat' arguments
for Linux and Windows in the SocketWIndowWordcount JavaDocs
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5c9caa7d2cc4d917898879ad9518c07e5cca5b49
Author: Jason <ji...@163.com>
AuthorDate: Mon Jun 3 14:25:30 2019 +0800
[hotfix] [examples] Distinguish 'netcat' arguments for Linux and Windows in the SocketWIndowWordcount JavaDocs
This closes #8593.
---
.../apache/flink/streaming/examples/socket/SocketWindowWordCount.java | 2 +-
.../flink/streaming/scala/examples/socket/SocketWindowWordCount.scala | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/socket/SocketWindowWordCount.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/socket/SocketWindowWordCount.java
index 646a74a..921ab3a 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/socket/SocketWindowWordCount.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/socket/SocketWindowWordCount.java
@@ -33,7 +33,7 @@ import org.apache.flink.util.Collector;
* The easiest way to try this out is to open a text server (at port 12345)
* using the <i>netcat</i> tool via
* <pre>
- * nc -l 12345
+ * nc -l 12345 on Linux or nc -l -p 12345 on Windows
* </pre>
* and run this example with the hostname and the port as arguments.
*/
diff --git a/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketWindowWordCount.scala b/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketWindowWordCount.scala
index e607f61..bdb1561 100644
--- a/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketWindowWordCount.scala
+++ b/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketWindowWordCount.scala
@@ -29,7 +29,7 @@ import org.apache.flink.streaming.api.windowing.time.Time
* The easiest way to try this out is to open a text sever (at port 12345)
* using the ''netcat'' tool via
* {{{
- * nc -l 12345
+ * nc -l 12345 on Linux or nc -l -p 12345 on Windows
* }}}
* and run this example with the hostname and the port as arguments..
*/