[BAHIR-91] Upgrade Flink version to 1.2.0

apache · Mar 1, 2017 · d1e0466 · d1e0466
1 parent a830077
commit d1e0466
Show file tree

Hide file tree

Showing 5 changed files with 48 additions and 6 deletions.
diff --git a/flink-connector-activemq/pom.xml b/flink-connector-activemq/pom.xml
@@ -85,6 +85,7 @@ under the License.
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
+
         <dependency>
             <groupId>org.apache.activemq.tooling</groupId>
             <artifactId>activemq-junit</artifactId>

diff --git a/...-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSourceTest.java b/...-activemq/src/test/java/org/apache/flink/streaming/connectors/activemq/AMQSourceTest.java
@@ -19,7 +19,10 @@
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.KeyedStateStore;
+import org.apache.flink.api.common.state.OperatorStateStore;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
 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;
@@ -97,6 +100,22 @@ public void before() throws Exception {
         amqSource = new AMQSource<>(config);
         amqSource.setRuntimeContext(createRuntimeContext());
         amqSource.open(new Configuration());
+        amqSource.initializeState(new FunctionInitializationContext() {
+            @Override
+            public boolean isRestored() {
+                return false;
+            }
+
+            @Override
+            public OperatorStateStore getOperatorStateStore() {
+                return mock(OperatorStateStore.class);
+            }
+
+            @Override
+            public KeyedStateStore getKeyedStateStore() {
+                return mock(KeyedStateStore.class);
+            }
+        });
     }
 
     private RuntimeContext createRuntimeContext() {

diff --git a/...src/test/java/org/apache/flink/streaming/connectors/activemq/ActiveMQConnectorITCase.java b/...src/test/java/org/apache/flink/streaming/connectors/activemq/ActiveMQConnectorITCase.java
@@ -22,14 +22,15 @@
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 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.ForkableFlinkMiniCluster;
 import org.apache.flink.test.util.SuccessException;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -51,7 +52,7 @@ public class ActiveMQConnectorITCase {
     public static final int MESSAGES_NUM = 10000;
     public static final String QUEUE_NAME = "queue";
     public static final String TOPIC_NAME = "topic";
-    private static ForkableFlinkMiniCluster flink;
+    private static LocalFlinkMiniCluster flink;
     private static int flinkPort;
 
     @BeforeClass
@@ -63,7 +64,7 @@ public static void beforeClass() {
         flinkConfig.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 16);
         flinkConfig.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "0 s");
 
-        flink = new ForkableFlinkMiniCluster(flinkConfig, false);
+        flink = new LocalFlinkMiniCluster(flinkConfig, false);
         flink.start();
 
         flinkPort = flink.getLeaderRPCPort();
@@ -211,9 +212,19 @@ public void run() {
         while (deadline.hasTimeLeft() && sourceContext.getIdsNum() < MESSAGES_NUM) {
             Thread.sleep(100);
             Random random = new Random();
-            long checkpointId = random.nextLong();
+            final long checkpointId = random.nextLong();
             synchronized (sourceContext.getCheckpointLock()) {
-                source.snapshotState(checkpointId, System.currentTimeMillis());
+                source.snapshotState(new FunctionSnapshotContext() {
+                    @Override
+                    public long getCheckpointId() {
+                        return checkpointId;
+                    }
+
+                    @Override
+                    public long getCheckpointTimestamp() {
+                        return System.currentTimeMillis();
+                    }
+                });
                 source.notifyCheckpointComplete(checkpointId);
             }
         }

diff --git a/flink-connector-redis/pom.xml b/flink-connector-redis/pom.xml
@@ -74,4 +74,15 @@ under the License.
             <scope>test</scope>
         </dependency>
     </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.felix</groupId>
+                <artifactId>maven-bundle-plugin</artifactId>
+                <inherited>true</inherited>
+                <extensions>true</extensions>
+            </plugin>
+        </plugins>
+    </build>
 </project>
diff --git a/pom.xml b/pom.xml
@@ -90,7 +90,7 @@
     <log4j.version>1.2.17</log4j.version>
 
     <!-- Spark version -->
-    <flink.version>1.1.1</flink.version>
+    <flink.version>1.2.0</flink.version>
 
     <PermGen>64m</PermGen>
     <MaxPermGen>512m</MaxPermGen>