[FLINK-7769][QS] Move queryable state outside the runtime.

Creates a separate for the queryable state and moves the client code outside the runtime. The Task Manager is now instantiating the KvStateServer using reflection.
apache · Oct 11, 2017 · 29a6e99 · 29a6e99
1 parent bc4638a
commit 29a6e99
Show file tree

Hide file tree

Showing 57 changed files with 1,507 additions and 1,154 deletions.
diff --git a/...-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java b/...-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
@@ -27,7 +27,7 @@
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
-import org.apache.flink.runtime.query.netty.message.KvStateRequestSerializer;
+import org.apache.flink.runtime.query.netty.message.KvStateSerializer;
 import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
 import org.apache.flink.runtime.state.internal.InternalKvState;
 import org.apache.flink.util.Preconditions;
@@ -125,8 +125,8 @@ public void setCurrentNamespace(N namespace) {
 	public byte[] getSerializedValue(byte[] serializedKeyAndNamespace) throws Exception {
 		Preconditions.checkNotNull(serializedKeyAndNamespace, "Serialized key and namespace");
 
-		//TODO make KvStateRequestSerializer key-group aware to save this round trip and key-group computation
+		//TODO make KvStateSerializer key-group aware to save this round trip and key-group computation
-		Tuple2<K, N> des = KvStateRequestSerializer.<K, N>deserializeKeyAndNamespace(
+		Tuple2<K, N> des = KvStateSerializer.<K, N>deserializeKeyAndNamespace(
 				serializedKeyAndNamespace,
 				backend.getKeySerializer(),
 				namespaceSerializer);

diff --git a/...ckend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java b/...ckend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBMapState.java
@@ -26,7 +26,7 @@
 import org.apache.flink.core.memory.ByteArrayOutputStreamWithPos;
 import org.apache.flink.core.memory.DataInputViewStreamWrapper;
 import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
-import org.apache.flink.runtime.query.netty.message.KvStateRequestSerializer;
+import org.apache.flink.runtime.query.netty.message.KvStateSerializer;
 import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
 import org.apache.flink.runtime.state.internal.InternalMapState;
 import org.apache.flink.util.Preconditions;
@@ -223,8 +223,8 @@ public void clear() {
 	public byte[] getSerializedValue(byte[] serializedKeyAndNamespace) throws Exception {
 		Preconditions.checkNotNull(serializedKeyAndNamespace, "Serialized key and namespace");
 
-		//TODO make KvStateRequestSerializer key-group aware to save this round trip and key-group computation
+		//TODO make KvStateSerializer key-group aware to save this round trip and key-group computation
-		Tuple2<K, N> des = KvStateRequestSerializer.deserializeKeyAndNamespace(
+		Tuple2<K, N> des = KvStateSerializer.deserializeKeyAndNamespace(
 				serializedKeyAndNamespace,
 				backend.getKeySerializer(),
 				namespaceSerializer);
@@ -248,7 +248,7 @@ public Map.Entry<UK, UV> next() {
 			return null;
 		}
 
-		return KvStateRequestSerializer.serializeMap(new Iterable<Map.Entry<UK, UV>>() {
+		return KvStateSerializer.serializeMap(new Iterable<Map.Entry<UK, UV>>() {
 			@Override
 			public Iterator<Map.Entry<UK, UV>> iterator() {
 				return iterator;

diff --git a/flink-queryable-state/flink-queryable-state-java/pom.xml b/flink-queryable-state/flink-queryable-state-java/pom.xml
@@ -0,0 +1,143 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.flink</groupId>
+        <artifactId>flink-queryable-state</artifactId>
+        <version>1.4-SNAPSHOT</version>
+        <relativePath>..</relativePath>
+    </parent>
+
+    <artifactId>flink-queryable-state-java_${scala.binary.version}</artifactId>
+    <name>flink-queryable-state-java</name>
+    <packaging>jar</packaging>
+
+    <dependencies>
+
+        <!-- core dependencies -->
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- test dependencies -->
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+       <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- ===================================================
+								Testing
+			=================================================== -->
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils-junit</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>${curator.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.data-artisans</groupId>
+            <artifactId>flakka-testkit_${scala.binary.version}</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>
diff --git a/...link/runtime/query/UnknownJobManager.java → ...ink/queryablestate/UnknownJobManager.java b/...link/runtime/query/UnknownJobManager.java → ...ink/queryablestate/UnknownJobManager.java
@@ -16,13 +16,15 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.query;
+package org.apache.flink.queryablestate;
+
+import org.apache.flink.queryablestate.client.KvStateLocationLookupService;
 
 /**
  * Exception to fail Future with if no JobManager is currently registered at
  * the {@link KvStateLocationLookupService}.
  */
-class UnknownJobManager extends Exception {
+public class UnknownJobManager extends Exception {
 
 	private static final long serialVersionUID = 1L;
 

diff --git a/...me/query/netty/UnknownKeyOrNamespace.java → ...queryablestate/UnknownKeyOrNamespace.java b/...me/query/netty/UnknownKeyOrNamespace.java → ...queryablestate/UnknownKeyOrNamespace.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.query.netty;
+package org.apache.flink.queryablestate;
 
 /**
  * Thrown if the KvState does not hold any state for the given key or namespace.
@@ -25,7 +25,7 @@ public class UnknownKeyOrNamespace extends IllegalStateException {
 
 	private static final long serialVersionUID = 1L;
 
-	UnknownKeyOrNamespace() {
+	public UnknownKeyOrNamespace() {
 		super("KvState does not hold any state for key/namespace.");
 	}
 }
diff --git a/...runtime/query/netty/UnknownKvStateID.java → ...link/queryablestate/UnknownKvStateID.java b/...runtime/query/netty/UnknownKvStateID.java → ...link/queryablestate/UnknownKvStateID.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.query.netty;
+package org.apache.flink.queryablestate;
 
 import org.apache.flink.runtime.query.KvStateID;
 import org.apache.flink.util.Preconditions;

diff --git a/...query/UnknownKvStateKeyGroupLocation.java → ...state/UnknownKvStateKeyGroupLocation.java b/...query/UnknownKvStateKeyGroupLocation.java → ...state/UnknownKvStateKeyGroupLocation.java
@@ -16,13 +16,15 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.query;
+package org.apache.flink.queryablestate;
+
+import org.apache.flink.runtime.query.KvStateLocation;
 
 /**
  * Exception thrown if there is no location information available for the given
  * key group in a {@link KvStateLocation} instance.
  */
-class UnknownKvStateKeyGroupLocation extends Exception {
+public class UnknownKvStateKeyGroupLocation extends Exception {
 
 	private static final long serialVersionUID = 1L;
 

diff --git a/...ery/AkkaKvStateLocationLookupService.java → ...ent/AkkaKvStateLocationLookupService.java b/...ery/AkkaKvStateLocationLookupService.java → ...ent/AkkaKvStateLocationLookupService.java
@@ -16,14 +16,17 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.query;
+package org.apache.flink.queryablestate.client;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.queryablestate.UnknownJobManager;
 import org.apache.flink.runtime.akka.AkkaUtils;
 import org.apache.flink.runtime.instance.ActorGateway;
 import org.apache.flink.runtime.instance.AkkaActorGateway;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.runtime.query.KvStateLocation;
+import org.apache.flink.runtime.query.KvStateMessage;
 import org.apache.flink.util.Preconditions;
 
 import akka.actor.ActorRef;
@@ -46,7 +49,7 @@
  * Akka-based {@link KvStateLocationLookupService} that retrieves the current
  * JobManager address and uses it for lookups.
  */
-class AkkaKvStateLocationLookupService implements KvStateLocationLookupService, LeaderRetrievalListener {
+public class AkkaKvStateLocationLookupService implements KvStateLocationLookupService, LeaderRetrievalListener {
 
 	private static final Logger LOG = LoggerFactory.getLogger(KvStateLocationLookupService.class);
 
@@ -76,7 +79,7 @@ class AkkaKvStateLocationLookupService implements KvStateLocationLookupService,
 	 * @param askTimeout             Timeout for JobManager ask-requests.
 	 * @param retryStrategyFactory   Retry strategy if no JobManager available.
 	 */
-	AkkaKvStateLocationLookupService(
+	public AkkaKvStateLocationLookupService(
 			LeaderRetrievalService leaderRetrievalService,
 			ActorSystem actorSystem,
 			FiniteDuration askTimeout,
@@ -209,7 +212,7 @@ public void handleError(Exception exception) {
 	 * }
 	 * </pre>
 	 */
-	interface LookupRetryStrategy {
+	public interface LookupRetryStrategy {
 
 		/**
 		 * Returns the current retry.
@@ -230,7 +233,7 @@ interface LookupRetryStrategy {
 	/**
 	 * Factory for retry strategies.
 	 */
-	interface LookupRetryStrategyFactory {
+	public interface LookupRetryStrategyFactory {
 
 		/**
 		 * Creates a new retry strategy.
@@ -244,7 +247,7 @@ interface LookupRetryStrategyFactory {
 	/**
 	 * Factory for disabled retries.
 	 */
-	static class DisabledLookupRetryStrategyFactory implements LookupRetryStrategyFactory {
+	public static class DisabledLookupRetryStrategyFactory implements LookupRetryStrategyFactory {
 
 		private static final DisabledLookupRetryStrategy RETRY_STRATEGY = new DisabledLookupRetryStrategy();
 
@@ -271,7 +274,7 @@ public boolean tryRetry() {
 	/**
 	 * Factory for fixed delay retries.
 	 */
-	static class FixedDelayLookupRetryStrategyFactory implements LookupRetryStrategyFactory {
+	public static class FixedDelayLookupRetryStrategyFactory implements LookupRetryStrategyFactory {
 
 		private final int maxRetries;
 		private final FiniteDuration retryDelay;

diff --git a/...nk/runtime/query/netty/KvStateClient.java → .../queryablestate/client/KvStateClient.java b/...nk/runtime/query/netty/KvStateClient.java → .../queryablestate/client/KvStateClient.java
@@ -16,12 +16,16 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.query.netty;
+package org.apache.flink.queryablestate.client;
 
+import org.apache.flink.queryablestate.UnknownKeyOrNamespace;
+import org.apache.flink.queryablestate.UnknownKvStateID;
+import org.apache.flink.queryablestate.network.messages.MessageSerializer;
 import org.apache.flink.runtime.io.network.netty.NettyBufferPool;
 import org.apache.flink.runtime.query.KvStateID;
+import org.apache.flink.runtime.query.KvStateServer;
 import org.apache.flink.runtime.query.KvStateServerAddress;
-import org.apache.flink.runtime.query.netty.message.KvStateRequestSerializer;
+import org.apache.flink.runtime.query.netty.KvStateRequestStats;
 import org.apache.flink.util.Preconditions;
 
 import org.apache.flink.shaded.guava18.com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -485,7 +489,7 @@ Future<byte[]> getKvState(KvStateID kvStateId, byte[] serializedKeyAndNamespace)
 
 				stats.reportRequest();
 
-				ByteBuf buf = KvStateRequestSerializer.serializeKvStateRequest(
+				ByteBuf buf = MessageSerializer.serializeKvStateRequest(
 						channel.alloc(),
 						requestId,
 						kvStateId,