feat: 添加对blink-1.5.1的支持

但是现在写state好像有问题，只好注释掉所有单元测试

.gitignore

@@ -5,3 +5,4 @@
 bin/
 build/
 gradle.properties
+.idea

bravo-test-utils/build.gradle

@@ -1,8 +1,8 @@
 
 dependencies {
-  compile group: 'org.apache.flink', name: 'flink-clients_2.11', version: versions.flink
-  compile group: 'org.apache.flink', name: 'flink-runtime_2.11', version: versions.flink
-  compile group: 'org.apache.flink', name: 'flink-test-utils_2.11', version: versions.flink
-  compile group: 'org.apache.flink', name: 'flink-statebackend-rocksdb_2.11', version: versions.flink
-  compile group: 'org.apache.flink', name: 'flink-streaming-java_2.11', version: versions.flink
+  compile group: 'com.alibaba.blink', name: 'flink-clients_2.11', version: versions.flink
+  compile group: 'com.alibaba.blink', name: 'flink-runtime_2.11', version: versions.flink
+  compile group: 'com.alibaba.blink', name: 'flink-test-utils_2.11', version: versions.flink
+  compile group: 'com.alibaba.blink', name: 'flink-statebackend-rocksdb_2.11', version: versions.flink
+  compile group: 'com.alibaba.blink', name: 'flink-streaming-java_2.11', version: versions.flink
 }

bravo-test-utils/src/main/java/com/king/bravo/testing/CollectingSink.java

@@ -20,10 +20,10 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
 import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 
-import com.google.common.collect.Lists;
 
 public class CollectingSink implements SinkFunction<String>, ListCheckpointed<Integer> {
 	private static final long serialVersionUID = 1L;
@@ -33,10 +33,11 @@ public class CollectingSink implements SinkFunction<String>, ListCheckpointed<In
 	public CollectingSink() {}
 
 	@Override
-	public void invoke(String out) throws Exception {
-		OUTPUT.add(out);
+	public void invoke(String value, Context context) throws Exception {
+		OUTPUT.add(value);
 	}
 
+
 	@Override
 	public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
 		return Lists.newArrayList(OUTPUT.size());

bravo-test-utils/src/main/java/com/king/bravo/testing/actions/OneTimePipelineAction.java

@@ -28,6 +28,7 @@ public abstract class OneTimePipelineAction implements PipelineAction {
 	private boolean checkpointLockTriggered = false;
 	private boolean clusterActionTriggered = false;
 
+	@Override
 	public final void withCheckpointLock(SourceContext<String> ctx) {
 		if (checkpointLockTriggered) {
 			return;
@@ -39,6 +40,7 @@ public final void withCheckpointLock(SourceContext<String> ctx) {
 
 	protected void onceWithCheckpointLock(SourceContext<String> ctx) {}
 
+	@Override
 	public final void executeClusterAction(ClusterClient<?> client, JobID id) throws Exception {
 
 		if (clusterActionTriggered) {

bravo/build.gradle

@@ -1,7 +1,7 @@
 
 dependencies {
   compile group: 'com.google.guava', name: 'guava', version: '18.0'
-  compileOnly group: 'org.apache.flink', name: 'flink-streaming-java_2.11', version: versions.flink
-  compileOnly group: 'org.apache.flink', name: 'flink-statebackend-rocksdb_2.11', version: versions.flink
+  compileOnly group: 'com.alibaba.blink', name: 'flink-streaming-java_2.11', version: versions.flink
+  compileOnly group: 'com.alibaba.blink', name: 'flink-statebackend-rocksdb_2.11', version: versions.flink
   testCompile (project(':bravo-test-utils'))
 }

bravo/src/main/java/com/king/bravo/reader/KeyedStateReader.java

@@ -100,8 +100,8 @@ public void configure(int maxParallelism, TypeSerializer<?> keySerializer, TypeS
 		}
 
 		if (this.valueDeserializer == null && this.valueDeserializerType == null) {
-			this.valueDeserializer = ttlSerializer ? StateMetadataUtils.unwrapTtlSerializer(valueSerializer)
-					: (TypeSerializer<V>) valueSerializer;
+			// ttl is valida from flink 1.6.0
+			this.valueDeserializer = (TypeSerializer<V>) valueSerializer;
 		}
 		initialized = true;
 	}

bravo/src/main/java/com/king/bravo/reader/OperatorStateReader.java

@@ -45,7 +45,6 @@
 import com.king.bravo.reader.inputformat.RocksDBKeyedStateInputFormat;
 import com.king.bravo.types.KeyedStateRow;
 import com.king.bravo.utils.StateMetadataUtils;
-import com.king.bravo.writer.OperatorStateWriter;
 
 /**
  * Utility for reading the states stored in a Flink operator into DataSets.
@@ -108,7 +107,7 @@ public <K, V, O> DataSet<O> readKeyedStates(KeyedStateReader<K, V, O> reader) th
 	}
 
 	private TypeSerializer<?> getKeySerializer(KeyedBackendSerializationProxy<?> proxy) {
-		TypeSerializer<?> keySerializer = proxy.getKeySerializerConfigSnapshot().restoreSerializer();
+		TypeSerializer<?> keySerializer = proxy.getKeySerializer();
 		if (keySerializer instanceof TupleSerializerBase) {
 			TupleSerializerBase ts = (TupleSerializerBase) keySerializer;
 			if (ts.getTupleClass().equals(Tuple1.class)) {
@@ -131,7 +130,6 @@ public DataSet<KeyedStateRow> getAllKeyedStateRows() {
 	 * Return all the keyed state rows that were not accessed using a reader.
 	 * This is a convenience method so we can union the untouched part of the
 	 * state with the changed parts before writing them back using the
-	 * {@link OperatorStateWriter}.
 	 */
 	public DataSet<KeyedStateRow> getAllUnreadKeyedStateRows() {
 		readKeyedStates();

bravo/src/main/java/com/king/bravo/reader/inputformat/RocksDBCheckpointIterator.java

@@ -40,10 +40,10 @@
 import org.apache.flink.runtime.state.IncrementalKeyedStateHandle;
 import org.apache.flink.runtime.state.StateHandleID;
 import org.apache.flink.runtime.state.StreamStateHandle;
-import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
 import org.apache.flink.util.FileUtils;
 import org.apache.flink.util.IOUtils;
 import org.apache.flink.util.Preconditions;
+import org.apache.flink.runtime.state.RegisteredKeyedBackendStateMetaInfo.Snapshot;
 import org.rocksdb.ColumnFamilyDescriptor;
 import org.rocksdb.ColumnFamilyHandle;
 import org.rocksdb.ColumnFamilyOptions;
@@ -74,7 +74,7 @@ public RocksDBCheckpointIterator(IncrementalKeyedStateHandle handle, FilterFunct
 			String localPath) {
 		this.localPath = localPath;
 		this.cancelStreamRegistry = new CloseableRegistry();
-		List<StateMetaInfoSnapshot> stateMetaInfoSnapshots = StateMetadataUtils
+		List<Snapshot<?,?>> stateMetaInfoSnapshots = StateMetadataUtils
 				.getKeyedBackendSerializationProxy(handle.getMetaStateHandle()).getStateMetaInfoSnapshots();
 
 		stateColumnFamilyHandles = new ArrayList<>(stateMetaInfoSnapshots.size() + 1);
@@ -148,7 +148,7 @@ private void copyStateDataHandleData(
 	}
 
 	private void createColumnIterators(FilterFunction<String> stateFilter,
-			List<StateMetaInfoSnapshot> stateMetaInfoSnapshots)
+			List<Snapshot<?,?>> stateMetaInfoSnapshots)
 			throws Exception {
 		Map<String, RocksIteratorWrapper> iterators = new HashMap<>();
 		for (int i = 0; i < stateMetaInfoSnapshots.size(); i++) {
@@ -179,11 +179,11 @@ private void updateCurrentIterator() {
 	}
 
 	private List<ColumnFamilyDescriptor> createAndRegisterColumnFamilyDescriptors(
-			List<StateMetaInfoSnapshot> stateMetaInfoSnapshots) {
+			List<Snapshot<?,?>> stateMetaInfoSnapshots) {
 
 		List<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>(stateMetaInfoSnapshots.size());
 
-		for (StateMetaInfoSnapshot stateMetaInfoSnapshot : stateMetaInfoSnapshots) {
+		for (Snapshot stateMetaInfoSnapshot : stateMetaInfoSnapshots) {
 			ColumnFamilyDescriptor columnFamilyDescriptor = new ColumnFamilyDescriptor(
 					stateMetaInfoSnapshot.getName().getBytes(ConfigConstants.DEFAULT_CHARSET),
 					colOptions);

bravo/src/main/java/com/king/bravo/utils/StateMetadataUtils.java

@@ -17,7 +17,6 @@
  */
 package com.king.bravo.utils;
 
-import org.apache.flink.api.common.typeutils.CompositeSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.core.fs.FSDataInputStream;
 import org.apache.flink.core.fs.FileSystem.WriteMode;
@@ -28,21 +27,12 @@
 import org.apache.flink.runtime.checkpoint.savepoint.Savepoint;
 import org.apache.flink.runtime.checkpoint.savepoint.SavepointV2;
 import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
-import org.apache.flink.runtime.state.IncrementalKeyedStateHandle;
-import org.apache.flink.runtime.state.KeyedBackendSerializationProxy;
-import org.apache.flink.runtime.state.KeyedStateHandle;
-import org.apache.flink.runtime.state.SnappyStreamCompressionDecorator;
-import org.apache.flink.runtime.state.StreamCompressionDecorator;
-import org.apache.flink.runtime.state.StreamStateHandle;
-import org.apache.flink.runtime.state.UncompressedStreamCompressionDecorator;
+import org.apache.flink.runtime.state.*;
 import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorage;
-import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
-import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot.CommonSerializerKeys;
+import org.apache.flink.runtime.state.RegisteredKeyedBackendStateMetaInfo.Snapshot;
 
 import java.io.DataInputStream;
 import java.io.IOException;
-import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.util.Arrays;
 import java.util.Collection;
@@ -126,8 +116,7 @@ public static Savepoint createNewSavepoint(Savepoint oldSavepoint, Collection<Op
 
 	public static Optional<KeyedBackendSerializationProxy<?>> getKeyedBackendSerializationProxy(OperatorState opState) {
 		try {
-			KeyedStateHandle firstHandle = opState.getStates().iterator().next().getManagedKeyedState().iterator()
-					.next();
+			KeyedStateHandle firstHandle = opState.getStates().iterator().next().getManagedKeyedState().iterator().next();
 			if (firstHandle instanceof IncrementalKeyedStateHandle) {
 				return Optional.of(getKeyedBackendSerializationProxy(
 						((IncrementalKeyedStateHandle) firstHandle).getMetaStateHandle()));
@@ -149,12 +138,12 @@ public static StreamCompressionDecorator getCompressionDecorator(KeyedBackendSer
 	public static <T> Optional<TypeSerializer<T>> getSerializer(KeyedBackendSerializationProxy<?> proxy,
 			String stateName) {
 
-		for (StateMetaInfoSnapshot snapshot : proxy.getStateMetaInfoSnapshots()) {
+		for (Snapshot snapshot : proxy.getStateMetaInfoSnapshots()) {
 			if (snapshot.getName().equals(stateName)) {
 				return Optional
 						.of((TypeSerializer<T>) snapshot
-								.getTypeSerializerSnapshot(CommonSerializerKeys.VALUE_SERIALIZER)
-								.restoreSerializer());
+                                .getStateSerializer()
+						);
 			}
 		}
 
@@ -165,7 +154,7 @@ public static Map<Integer, String> getStateIdMapping(KeyedBackendSerializationPr
 		Map<Integer, String> stateIdMapping = new HashMap<>();
 
 		int stateId = 0;
-		for (StateMetaInfoSnapshot snapshot : proxy.getStateMetaInfoSnapshots()) {
+		for (Snapshot snapshot : proxy.getStateMetaInfoSnapshots()) {
 			stateIdMapping.put(stateId, snapshot.getName());
 			stateId++;
 		}
@@ -176,7 +165,7 @@ public static Map<Integer, String> getStateIdMapping(KeyedBackendSerializationPr
 	public static KeyedBackendSerializationProxy<?> getKeyedBackendSerializationProxy(
 			StreamStateHandle streamStateHandle) {
 		KeyedBackendSerializationProxy<Integer> serializationProxy = new KeyedBackendSerializationProxy<>(
-				StateMetadataUtils.class.getClassLoader());
+				StateMetadataUtils.class.getClassLoader(),false);
 		try (FSDataInputStream is = streamStateHandle.openInputStream()) {
 			DataInputViewStreamWrapper iw = new DataInputViewStreamWrapper(is);
 			serializationProxy.read(iw);
@@ -199,10 +188,11 @@ public static boolean isTtlState(TypeSerializer<?> valueSerializer) {
 		return ttlSerializer;
 	}
 
-	public static <T> TypeSerializer<T> unwrapTtlSerializer(TypeSerializer<?> valueSerializer) throws Exception {
-		Field f = CompositeSerializer.class.getDeclaredField("fieldSerializers");
-		f.setAccessible(true);
-		return (TypeSerializer<T>) ((TypeSerializer<Object>[]) f.get(valueSerializer))[1];
-	}
+	// ttl is valida from flink 1.6.0
+//	public static <T> TypeSerializer<T> unwrapTtlSerializer(TypeSerializer<?> valueSerializer) throws Exception {
+//		Field f = CompositeSerializer.class.getDeclaredField("fieldSerializers");
+//		f.setAccessible(true);
+//		return (TypeSerializer<T>) ((TypeSerializer<Object>[]) f.get(valueSerializer))[1];
+//	}
 
 }

bravo/src/main/java/com/king/bravo/writer/OperatorStateWriter.java

@@ -29,7 +29,6 @@
 import org.apache.flink.api.common.operators.Order;
 import org.apache.flink.api.common.state.StateDescriptor;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.core.fs.Path;
@@ -39,17 +38,9 @@
 import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
 import org.apache.flink.runtime.checkpoint.StateObjectCollection;
 import org.apache.flink.runtime.checkpoint.savepoint.Savepoint;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
-import org.apache.flink.runtime.state.CheckpointedStateScope;
-import org.apache.flink.runtime.state.KeyedBackendSerializationProxy;
-import org.apache.flink.runtime.state.KeyedStateHandle;
-import org.apache.flink.runtime.state.OperatorStateBackend;
-import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo;
-import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.runtime.state.*;
 import org.apache.flink.runtime.state.filesystem.FileBasedStateOutputStream;
-import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
-import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
+import org.apache.flink.runtime.state.RegisteredKeyedBackendStateMetaInfo.Snapshot;
 
 import com.google.common.collect.HashBiMap;
 import com.google.common.collect.Lists;
@@ -74,7 +65,7 @@ public class OperatorStateWriter {
 	private Path newCheckpointBasePath;
 	private long checkpointId;
 	private BiConsumer<Integer, OperatorStateBackend> transformer;
-	private Map<String, StateMetaInfoSnapshot> metaSnapshots;
+	private Map<String, Snapshot> metaSnapshots;
 	private KeyedBackendSerializationProxy<?> proxy;
 
 	private boolean keepBaseKeyedStates = true;
@@ -94,19 +85,15 @@ public OperatorStateWriter(long checkpointId, OperatorState baseOpState, Path ne
 		metaSnapshots = new HashMap<>();
 		if (proxy != null) {
 			proxy.getStateMetaInfoSnapshots()
-					.forEach(ms -> metaSnapshots.put(ms.getName(),
-							new StateMetaInfoSnapshot(ms.getName(), ms.getBackendStateType(), ms.getOptionsImmutable(),
-									ms.getSerializerSnapshotsImmutable(),
-									Maps.transformValues(ms.getSerializerSnapshotsImmutable(),
-											TypeSerializerSnapshot::restoreSerializer))));
+					.forEach(ms -> metaSnapshots.put(ms.getName(), ms));
 		}
 	}
 
 	/**
 	 * Defines the keyserializer for this operator. This method can be used when
 	 * adding state to a previously stateless operator where the keyserializer
 	 * is not available from the state.
-	 * 
+	 *
 	 * @param keySerializer
 	 */
 	public void setKeySerializer(TypeSerializer<?> keySerializer) {
@@ -120,7 +107,7 @@ public void setKeySerializer(TypeSerializer<?> keySerializer) {
 	 * <p>
 	 * This can be used to add all different kinds of keyed states: value, list,
 	 * map
-	 * 
+	 *
 	 * @param rows
 	 *            State rows to be added
 	 */
@@ -131,7 +118,7 @@ public void addKeyedStateRows(DataSet<KeyedStateRow> rows) {
 
 	/**
 	 * Removes the state metadata and rows for the given statename.
-	 * 
+	 *
 	 * @param stateName
 	 *            Name of the state to be deleted
 	 */
@@ -146,15 +133,15 @@ private void updateProxy() {
 					"KeySerializer must be defined when adding state to a previously stateless operator. Use writer.setKeySerializer(...)");
 		}
 
-		proxy = new KeyedBackendSerializationProxy<>(
+		proxy = new KeyedBackendSerializationProxy(
 				getKeySerializer(),
 				new ArrayList<>(metaSnapshots.values()),
 				proxy != null ? proxy.isUsingKeyGroupCompression() : true);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private <T> TypeSerializer<T> getKeySerializer() {
-		return proxy != null ? (TypeSerializer) proxy.getKeySerializerConfigSnapshot().restoreSerializer()
+		return proxy != null ? (TypeSerializer) proxy.getKeySerializer()
 				: (TypeSerializer) keySerializer;
 	}
 
@@ -168,7 +155,7 @@ private <T> TypeSerializer<T> getKeySerializer() {
 	 * <p>
 	 * Keep in mind that any state rows for the same name already added (through
 	 * {@link #addKeyedStateRows(DataSet)}) will not be overwritten.
-	 * 
+	 *
 	 * @param stateName
 	 * @param newState
 	 */
@@ -196,15 +183,15 @@ public <K, V> void addValueState(String stateName, DataSet<Tuple2<K, V>> newStat
 	 * <p>
 	 * When redefining a pre-existing state make sure you haven't added that as
 	 * keyed state rows before.
-	 * 
+	 *
 	 * @param stateName
 	 * @param newState
 	 * @param valueSerializer
 	 */
 	public <K, V> void createNewValueState(String stateName, DataSet<Tuple2<K, V>> newState,
 			TypeSerializer<V> valueSerializer) {
 
-		metaSnapshots.put(stateName, new RegisteredKeyValueStateBackendMetaInfo<>(StateDescriptor.Type.VALUE, stateName,
+		metaSnapshots.put(stateName, new RegisteredKeyedBackendStateMetaInfo<>(StateDescriptor.Type.VALUE, stateName,
 				VoidNamespaceSerializer.INSTANCE, valueSerializer).snapshot());
 
 		updateProxy();
@@ -218,7 +205,7 @@ public <K, V> void createNewValueState(String stateName, DataSet<Tuple2<K, V>> n
 	/**
 	 * Triggers the batch processing operations to write the operator state data
 	 * to persistent storage and create the metadata object
-	 * 
+	 *
 	 * @return {@link OperatorState} metadata pointing to the newly written
 	 *         state
 	 */
@@ -298,9 +285,8 @@ private StateObjectCollection<OperatorStateHandle> transformSubtaskOpState(Path
 			OperatorStateHandle newSnapshot = opBackend
 					.snapshot(checkpointId, System.currentTimeMillis(), new CheckpointStreamFactory() {
 						@Override
-						public CheckpointStateOutputStream createCheckpointStateOutputStream(
-								CheckpointedStateScope scope)
-								throws IOException {
+						public CheckpointStateOutputStream
+						createCheckpointStateOutputStream(long checkpointId, CheckpointedStateScope scope) throws IOException {
 							return new FileBasedStateOutputStream(outDir.getFileSystem(),
 									new Path(outDir, String.valueOf(UUID.randomUUID())));
 						}
@@ -319,7 +305,7 @@ public CheckpointStateOutputStream createCheckpointStateOutputStream(
 	 * <p>
 	 * The transformation will be executed sequentially, in-memory on the
 	 * client.
-	 * 
+	 *
 	 * @param transformer
 	 *            Consumer to be applied on the {@link OperatorStateBackend}
 	 * @throws Exception