You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bahir.apache.org by lr...@apache.org on 2018/07/26 13:36:30 UTC
bahir-flink git commit: [BAHIR-173] Update Flink version to 1.5.1
Repository: bahir-flink
Updated Branches:
refs/heads/master 55605cbde -> 567792b26
[BAHIR-173] Update Flink version to 1.5.1
Closes #29
Project: http://git-wip-us.apache.org/repos/asf/bahir-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/bahir-flink/commit/567792b2
Tree: http://git-wip-us.apache.org/repos/asf/bahir-flink/tree/567792b2
Diff: http://git-wip-us.apache.org/repos/asf/bahir-flink/diff/567792b2
Branch: refs/heads/master
Commit: 567792b26f5b9d80cb423f5d660e3011d475057c
Parents: 55605cb
Author: Dominik.Wosinski <bl...@wp.pl>
Authored: Fri Jul 20 23:47:12 2018 +0200
Committer: Luciano Resende <lr...@apache.org>
Committed: Thu Jul 26 09:35:20 2018 -0400
----------------------------------------------------------------------
.travis.yml | 12 ++-----
.../streaming/connectors/activemq/AMQSink.java | 4 +--
.../connectors/activemq/AMQSinkConfig.java | 2 +-
.../connectors/activemq/AMQSource.java | 6 ++--
.../connectors/activemq/AMQSourceConfig.java | 2 +-
.../connectors/activemq/AMQSinkTest.java | 10 +++---
.../connectors/activemq/AMQSourceTest.java | 14 ++++----
.../activemq/ActiveMQConnectorITCase.java | 13 +++----
flink-connector-akka/pom.xml | 7 +++-
.../streaming/connectors/akka/AkkaSource.java | 6 ++--
.../connectors/akka/AkkaSourceTest.java | 6 ++--
.../streaming/connectors/flume/FlumeSink.java | 5 ++-
.../connectors/influxdb/InfluxDBSink.java | 2 +-
.../connectors/netty/example/NettyUtil.scala | 3 --
.../netty/example/StreamSqlExample.scala | 4 +--
.../streaming/connectors/redis/RedisSink.java | 4 +--
.../connectors/redis/RedisITCaseBase.java | 4 +--
.../operator/StreamInMemOutputHandler.java | 4 +--
.../siddhi/operator/StreamOutputHandler.java | 4 +--
.../flink/streaming/siddhi/SiddhiCEPITCase.java | 4 +--
pom.xml | 36 +-------------------
21 files changed, 58 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index c3baf70..083e75d 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -32,20 +32,12 @@ matrix:
include:
- jdk: oraclejdk8
env:
- - FLINK_VERSION="1.3.0" SCALA_VERSION="2.11"
+ - FLINK_VERSION="1.5.1" SCALA_VERSION="2.11"
- CACHE_NAME=JDK8_F130_A
- - jdk: oraclejdk8
- env:
- - FLINK_VERSION="1.3.0" SCALA_VERSION="2.10"
- - CACHE_NAME=JDK8_F130_B
- jdk: openjdk8
env:
- - FLINK_VERSION="1.3.0" SCALA_VERSION="2.11"
+ - FLINK_VERSION="1.5.1" SCALA_VERSION="2.11"
- CACHE_NAME=JDK8_F130_C
- - jdk: openjdk8
- env:
- - FLINK_VERSION="1.3.0" SCALA_VERSION="2.10"
- - CACHE_NAME=JDK8_F130_D
before_install:
- ./dev/change-scala-version.sh $SCALA_VERSION
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSink.java
----------------------------------------------------------------------
diff --git a/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSink.java b/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSink.java
index a494162..b33bffe 100644
--- a/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSink.java
+++ b/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSink.java
@@ -18,10 +18,10 @@
package org.apache.flink.streaming.connectors.activemq;
import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.connectors.activemq.internal.AMQUtil;
-import org.apache.flink.streaming.util.serialization.SerializationSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -128,7 +128,7 @@ public class AMQSink<IN> extends RichSinkFunction<IN> {
* The incoming data
*/
@Override
- public void invoke(IN value) {
+ public void invoke(IN value, Context context) throws Exception {
try {
byte[] bytes = serializationSchema.serialize(value);
BytesMessage message = session.createBytesMessage();
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSinkConfig.java
----------------------------------------------------------------------
diff --git a/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSinkConfig.java b/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSinkConfig.java
index e10c3c8..51a6aac 100644
--- a/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSinkConfig.java
+++ b/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSinkConfig.java
@@ -17,7 +17,7 @@
package org.apache.flink.streaming.connectors.activemq;
import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.flink.streaming.util.serialization.SerializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.util.Preconditions;
/**
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSource.java
----------------------------------------------------------------------
diff --git a/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSource.java b/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSource.java
index e64b8fd..4f2114f 100644
--- a/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSource.java
+++ b/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSource.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.connectors.activemq;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
@@ -28,7 +29,6 @@ import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.connectors.activemq.internal.AMQExceptionListener;
import org.apache.flink.streaming.connectors.activemq.internal.AMQUtil;
import org.apache.flink.streaming.connectors.activemq.internal.RunningChecker;
-import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,7 +40,7 @@ import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import java.util.HashMap;
-import java.util.List;
+import java.util.Set;
/**
* Source for reading messages from an ActiveMQ queue.
@@ -191,7 +191,7 @@ public class AMQSource<OUT> extends MessageAcknowledgingSourceBase<OUT, String>
}
@Override
- protected void acknowledgeIDs(long checkpointId, List<String> UIds) {
+ protected void acknowledgeIDs(long checkpointId, Set<String> UIds) {
try {
for (String messageId : UIds) {
Message unacknowledgedMessage = unacknowledgedMessages.get(messageId);
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSourceConfig.java
----------------------------------------------------------------------
diff --git a/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSourceConfig.java b/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSourceConfig.java
index dd73b0e..06cb267 100644
--- a/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSourceConfig.java
+++ b/flink-connector-activemq/src/main/java/org/apache/flink/streaming/connectors/activemq/AMQSourceConfig.java
@@ -17,8 +17,8 @@
package org.apache.flink.streaming.connectors.activemq;
import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.streaming.connectors.activemq.internal.RunningChecker;
-import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.apache.flink.util.Preconditions;
/**
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSinkTest.java b/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSinkTest.java
index e4d67c3..52493d7 100644
--- a/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSinkTest.java
+++ b/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSinkTest.java
@@ -19,9 +19,9 @@
package org.apache.flink.streaming.connectors.activemq;
import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.util.serialization.SerializationSchema;
-import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.junit.Before;
import org.junit.Test;
@@ -83,7 +83,7 @@ public class AMQSinkTest {
@Test
public void messageSentToProducer() throws Exception {
byte[] expectedMessage = serializationSchema.serialize("msg");
- amqSink.invoke("msg");
+ amqSink.invoke("msg", null);
verify(producer).send(message);
verify(message).writeBytes(expectedMessage);
@@ -121,7 +121,7 @@ public class AMQSinkTest {
when(session.createBytesMessage()).thenThrow(JMSException.class);
amqSink.setLogFailuresOnly(true);
- amqSink.invoke("msg");
+ amqSink.invoke("msg", null);
}
@SuppressWarnings("unchecked")
@@ -129,7 +129,7 @@ public class AMQSinkTest {
public void exceptionOnSendAreThrownByDefault() throws Exception {
when(session.createBytesMessage()).thenThrow(JMSException.class);
- amqSink.invoke("msg");
+ amqSink.invoke("msg", null);
}
@Test
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSourceTest.java b/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSourceTest.java
index 0e6dd31..9c7be72 100644
--- a/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSourceTest.java
+++ b/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSourceTest.java
@@ -19,6 +19,7 @@ package org.apache.flink.streaming.connectors.activemq;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.KeyedStateStore;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.configuration.Configuration;
@@ -27,7 +28,6 @@ import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.connectors.activemq.internal.AMQExceptionListener;
import org.apache.flink.streaming.connectors.activemq.internal.RunningChecker;
-import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.junit.Before;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
@@ -161,7 +161,7 @@ public class AMQSourceTest {
@Test
public void acknowledgeReceivedMessage() throws Exception {
amqSource.run(context);
- amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singletonList(MSG_ID));
+ amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singleton(MSG_ID));
verify(message).acknowledge();
}
@@ -169,7 +169,7 @@ public class AMQSourceTest {
@Test
public void handleUnknownIds() throws Exception {
amqSource.run(context);
- amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singletonList("unknown-id"));
+ amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singleton("unknown-id"));
verify(message, never()).acknowledge();
}
@@ -177,8 +177,8 @@ public class AMQSourceTest {
@Test
public void doNotAcknowledgeMessageTwice() throws Exception {
amqSource.run(context);
- amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singletonList(MSG_ID));
- amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singletonList(MSG_ID));
+ amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singleton(MSG_ID));
+ amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singleton(MSG_ID));
verify(message, times(1)).acknowledge();
}
@@ -196,7 +196,7 @@ public class AMQSourceTest {
doThrow(JMSException.class).when(message).acknowledge();
amqSource.run(context);
- amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singletonList(MSG_ID));
+ amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singleton(MSG_ID));
}
@Test
@@ -206,7 +206,7 @@ public class AMQSourceTest {
doThrow(JMSException.class).when(message).acknowledge();
amqSource.run(context);
- amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singletonList(MSG_ID));
+ amqSource.acknowledgeIDs(CHECKPOINT_ID, Collections.singleton(MSG_ID));
}
@Test
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/ActiveMQConnectorITCase.java
----------------------------------------------------------------------
diff --git a/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/ActiveMQConnectorITCase.java b/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/ActiveMQConnectorITCase.java
index 9af6dd5..3b31fc4 100644
--- a/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/ActiveMQConnectorITCase.java
+++ b/flink-connector-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/ActiveMQConnectorITCase.java
@@ -20,8 +20,10 @@ package org.apache.flink.streaming.connectors.activemq;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
@@ -30,7 +32,6 @@ import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.test.util.SuccessException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -60,8 +61,8 @@ public class ActiveMQConnectorITCase {
// start also a re-usable Flink mini cluster
Configuration flinkConfig = new Configuration();
flinkConfig.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);
- flinkConfig.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 8);
- flinkConfig.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 16);
+ flinkConfig.setInteger(TaskManagerOptions.NUM_TASK_SLOTS.key(), 8);
+ flinkConfig.setInteger(TaskManagerOptions.MANAGED_MEMORY_SIZE.key(), 16);
flinkConfig.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "0 s");
flink = new LocalFlinkMiniCluster(flinkConfig, false);
@@ -74,7 +75,7 @@ public class ActiveMQConnectorITCase {
public static void afterClass() {
flinkPort = -1;
if (flink != null) {
- flink.shutdown();
+ flink.startInternalShutdown();
}
}
@@ -158,7 +159,7 @@ public class ActiveMQConnectorITCase {
.addSink(new SinkFunction<String>() {
final HashSet<Integer> set = new HashSet<>();
@Override
- public void invoke(String value) throws Exception {
+ public void invoke(String value, Context context) throws Exception {
int val = Integer.parseInt(value.split("-")[1]);
set.add(val);
@@ -181,7 +182,7 @@ public class ActiveMQConnectorITCase {
sink.open(new Configuration());
for (int i = 0; i < MESSAGES_NUM; i++) {
- sink.invoke("amq-" + i);
+ sink.invoke("amq-" + i, null);
}
AMQSourceConfig<String> sourceConfig = new AMQSourceConfig.AMQSourceConfigBuilder<String>()
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-akka/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connector-akka/pom.xml b/flink-connector-akka/pom.xml
index 5eb261b..3dab6b8 100644
--- a/flink-connector-akka/pom.xml
+++ b/flink-connector-akka/pom.xml
@@ -35,11 +35,16 @@ under the License.
<properties>
<mockito.version>1.10.19</mockito.version>
- <akka.version>2.3.15</akka.version>
+ <akka.version>2.4.20</akka.version>
</properties>
<dependencies>
<dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-akka/src/main/java/org/apache/flink/streaming/connectors/akka/AkkaSource.java
----------------------------------------------------------------------
diff --git a/flink-connector-akka/src/main/java/org/apache/flink/streaming/connectors/akka/AkkaSource.java b/flink-connector-akka/src/main/java/org/apache/flink/streaming/connectors/akka/AkkaSource.java
index 3925d0b..d8a6919 100644
--- a/flink-connector-akka/src/main/java/org/apache/flink/streaming/connectors/akka/AkkaSource.java
+++ b/flink-connector-akka/src/main/java/org/apache/flink/streaming/connectors/akka/AkkaSource.java
@@ -30,6 +30,8 @@ import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.akka.utils.ReceiverActor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import scala.concurrent.Await;
+import scala.concurrent.duration.Duration;
import java.util.Collections;
@@ -95,7 +97,7 @@ public class AkkaSource extends RichSourceFunction<Object>
Props.create(classForActor, ctx, urlOfPublisher, autoAck), actorName);
LOG.info("Started the Receiver actor {} successfully", actorName);
- receiverActorSystem.awaitTermination();
+ Await.result(receiverActorSystem.whenTerminated(), Duration.Inf());
}
@Override
@@ -103,7 +105,7 @@ public class AkkaSource extends RichSourceFunction<Object>
LOG.info("Closing source");
if (receiverActorSystem != null) {
receiverActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
- receiverActorSystem.shutdown();
+ receiverActorSystem.terminate();
}
}
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-akka/src/test/java/org/apache/flink/streaming/connectors/akka/AkkaSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connector-akka/src/test/java/org/apache/flink/streaming/connectors/akka/AkkaSourceTest.java b/flink-connector-akka/src/test/java/org/apache/flink/streaming/connectors/akka/AkkaSourceTest.java
index 99a1893..6327bdd 100644
--- a/flink-connector-akka/src/test/java/org/apache/flink/streaming/connectors/akka/AkkaSourceTest.java
+++ b/flink-connector-akka/src/test/java/org/apache/flink/streaming/connectors/akka/AkkaSourceTest.java
@@ -34,6 +34,8 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.duration.Duration;
import java.io.File;
import java.util.ArrayList;
@@ -80,8 +82,8 @@ public class AkkaSourceTest {
@After
public void afterTest() throws Exception {
- feederActorSystem.shutdown();
- feederActorSystem.awaitTermination();
+ feederActorSystem.terminate();
+ Await.result(feederActorSystem.whenTerminated(), Duration.Inf());
source.cancel();
sourceThread.join();
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java
----------------------------------------------------------------------
diff --git a/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java b/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java
index 00bfc39..41b1b25 100644
--- a/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java
+++ b/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java
@@ -17,10 +17,10 @@
package org.apache.flink.streaming.connectors.flume;
+import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
-import org.apache.flink.streaming.util.serialization.SerializationSchema;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.FlumeException;
@@ -55,8 +55,7 @@ public class FlumeSink<IN> extends RichSinkFunction<IN> {
* The tuple arriving from the datastream
*/
@Override
- public void invoke(IN value) {
-
+ public void invoke(IN value, Context context) throws Exception {
byte[] data = schema.serialize(value);
client.sendDataToFlume(data);
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-influxdb/src/main/java/org/apache/flink/streaming/connectors/influxdb/InfluxDBSink.java
----------------------------------------------------------------------
diff --git a/flink-connector-influxdb/src/main/java/org/apache/flink/streaming/connectors/influxdb/InfluxDBSink.java b/flink-connector-influxdb/src/main/java/org/apache/flink/streaming/connectors/influxdb/InfluxDBSink.java
index 03521b9..e7f8916 100644
--- a/flink-connector-influxdb/src/main/java/org/apache/flink/streaming/connectors/influxdb/InfluxDBSink.java
+++ b/flink-connector-influxdb/src/main/java/org/apache/flink/streaming/connectors/influxdb/InfluxDBSink.java
@@ -76,7 +76,7 @@ public class InfluxDBSink extends RichSinkFunction<InfluxDBPoint> {
* @param dataPoint {@link InfluxDBPoint}
*/
@Override
- public void invoke(InfluxDBPoint dataPoint) throws Exception {
+ public void invoke(InfluxDBPoint dataPoint, Context context) throws Exception {
if (StringUtils.isNullOrWhitespaceOnly(dataPoint.getMeasurement())) {
throw new RuntimeException("No measurement defined");
}
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-netty/src/main/scala/org/apache/flink/streaming/connectors/netty/example/NettyUtil.scala
----------------------------------------------------------------------
diff --git a/flink-connector-netty/src/main/scala/org/apache/flink/streaming/connectors/netty/example/NettyUtil.scala b/flink-connector-netty/src/main/scala/org/apache/flink/streaming/connectors/netty/example/NettyUtil.scala
index 56c551d..f6068bc 100644
--- a/flink-connector-netty/src/main/scala/org/apache/flink/streaming/connectors/netty/example/NettyUtil.scala
+++ b/flink-connector-netty/src/main/scala/org/apache/flink/streaming/connectors/netty/example/NettyUtil.scala
@@ -21,7 +21,6 @@ import java.io.{BufferedReader, InputStreamReader}
import java.net._
import org.apache.commons.lang3.SystemUtils
-import org.mortbay.util.MultiException
import org.slf4j.LoggerFactory
import scala.collection.JavaConverters._
@@ -153,8 +152,6 @@ object NettyUtil {
exception match {
case e: BindException if e.getMessage != null => true
case e: BindException => isBindCollision(e.getCause)
- case e: MultiException =>
- e.getThrowables.asScala.toList.map(_.asInstanceOf[Throwable]).exists(isBindCollision)
case e: Exception => isBindCollision(e.getCause)
case _ => false
}
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-netty/src/test/scala/org/apache/flink/streaming/connectors/netty/example/StreamSqlExample.scala
----------------------------------------------------------------------
diff --git a/flink-connector-netty/src/test/scala/org/apache/flink/streaming/connectors/netty/example/StreamSqlExample.scala b/flink-connector-netty/src/test/scala/org/apache/flink/streaming/connectors/netty/example/StreamSqlExample.scala
index ee634d0..5167b3e 100644
--- a/flink-connector-netty/src/test/scala/org/apache/flink/streaming/connectors/netty/example/StreamSqlExample.scala
+++ b/flink-connector-netty/src/test/scala/org/apache/flink/streaming/connectors/netty/example/StreamSqlExample.scala
@@ -60,9 +60,9 @@ object StreamSqlExample {
tEnv.registerDataStream("OrderA", orderA, 'user, 'product, 'amount)
// union the two tables
- val result = tEnv.sql("SELECT STREAM * FROM OrderA WHERE amount > 2")
+ val result = tEnv.sqlQuery("SELECT STREAM * FROM OrderA WHERE amount > 2")
- result.toDataStream[Order].print()
+ result.toAppendStream[Order].print()
env.execute()
}
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/RedisSink.java
----------------------------------------------------------------------
diff --git a/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/RedisSink.java b/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/RedisSink.java
index 9138862..6a03f11 100644
--- a/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/RedisSink.java
+++ b/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/RedisSink.java
@@ -26,8 +26,8 @@ import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSenti
import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainer;
import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainerBuilder;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
-import org.apache.flink.streaming.connectors.redis.common.mapper.RedisDataType;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
+import org.apache.flink.streaming.connectors.redis.common.mapper.RedisDataType;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.slf4j.Logger;
@@ -125,7 +125,7 @@ public class RedisSink<IN> extends RichSinkFunction<IN> {
* @param input The incoming data
*/
@Override
- public void invoke(IN input) throws Exception {
+ public void invoke(IN input, Context context) throws Exception {
String key = redisSinkMapper.getKeyFromData(input);
String value = redisSinkMapper.getValueFromData(input);
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisITCaseBase.java
----------------------------------------------------------------------
diff --git a/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisITCaseBase.java b/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisITCaseBase.java
index 18cdc64..2bbbcb7 100644
--- a/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisITCaseBase.java
+++ b/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisITCaseBase.java
@@ -16,7 +16,7 @@
*/
package org.apache.flink.streaming.connectors.redis;
-import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
+import org.apache.flink.test.util.AbstractTestBase;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import redis.embedded.RedisServer;
@@ -25,7 +25,7 @@ import java.io.IOException;
import static org.apache.flink.util.NetUtils.getAvailablePort;
-public abstract class RedisITCaseBase extends StreamingMultipleProgramsTestBase {
+public abstract class RedisITCaseBase extends AbstractTestBase {
public static final int REDIS_PORT = getAvailablePort();
public static final String REDIS_HOST = "127.0.0.1";
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamInMemOutputHandler.java
----------------------------------------------------------------------
diff --git a/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamInMemOutputHandler.java b/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamInMemOutputHandler.java
index 7af37ce..49c27c2 100755
--- a/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamInMemOutputHandler.java
+++ b/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamInMemOutputHandler.java
@@ -24,6 +24,8 @@ import java.util.Map;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.siddhi.utils.SiddhiTupleFactory;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.slf4j.Logger;
@@ -31,8 +33,6 @@ import org.slf4j.LoggerFactory;
import org.wso2.siddhi.core.event.Event;
import org.wso2.siddhi.core.stream.output.StreamCallback;
import org.wso2.siddhi.query.api.definition.AbstractDefinition;
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.ObjectMapper;
/**
* Siddhi Stream output callback handler and conver siddhi {@link Event} to required output type,
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamOutputHandler.java
----------------------------------------------------------------------
diff --git a/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamOutputHandler.java b/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamOutputHandler.java
index 8840dac..1c1096c 100755
--- a/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamOutputHandler.java
+++ b/flink-library-siddhi/src/main/java/org/apache/flink/streaming/siddhi/operator/StreamOutputHandler.java
@@ -17,11 +17,11 @@
package org.apache.flink.streaming.siddhi.operator;
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.siddhi.utils.SiddhiTupleFactory;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java
----------------------------------------------------------------------
diff --git a/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java b/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java
index 821c594..651288f 100755
--- a/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java
+++ b/flink-library-siddhi/src/test/java/org/apache/flink/streaming/siddhi/SiddhiCEPITCase.java
@@ -41,7 +41,7 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.operators.StreamMap;
-import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
+import org.apache.flink.test.util.AbstractTestBase;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -52,7 +52,7 @@ import static org.junit.Assert.assertEquals;
/**
* Flink-siddhi library integration test cases
*/
-public class SiddhiCEPITCase extends StreamingMultipleProgramsTestBase implements Serializable {
+public class SiddhiCEPITCase extends AbstractTestBase implements Serializable {
@Rule
public transient TemporaryFolder tempFolder = new TemporaryFolder();
http://git-wip-us.apache.org/repos/asf/bahir-flink/blob/567792b2/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index c2d36ed..d2d435c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -93,7 +93,7 @@
<log4j.version>1.2.17</log4j.version>
<!-- Flink version -->
- <flink.version>1.3.0</flink.version>
+ <flink.version>1.5.1</flink.version>
<PermGen>64m</PermGen>
<MaxPermGen>512m</MaxPermGen>
@@ -704,40 +704,6 @@
<module>distribution</module>
</modules>
</profile>
-
- <profile>
- <id>scala-2.10</id>
- <properties>
- <scala.version>2.10.5</scala.version>
- <scala.binary.version>2.10</scala.binary.version>
- </properties>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-enforcer-plugin</artifactId>
- <executions>
- <execution>
- <id>enforce-versions</id>
- <goals>
- <goal>enforce</goal>
- </goals>
- <configuration>
- <rules>
- <bannedDependencies>
- <excludes combine.children="append">
- <exclude>*:*_2.11</exclude>
- </excludes>
- </bannedDependencies>
- </rules>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
- </profile>
-
<profile>
<id>scala-2.11</id>
<activation>