[flink] branch master updated: [FLINK-8995][tests] Add keyed state that uses custom stateful serializer to allround test job

[sr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

srichter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 0e9a249  [FLINK-8995][tests] Add keyed state that uses custom stateful serializer to allround test job
0e9a249 is described below

commit 0e9a249fb8953735d6af93795e95e2abec223626
Author: Stefan Richter <s....@data-artisans.com>
AuthorDate: Tue Oct 23 17:15:26 2018 +0200

    [FLINK-8995][tests] Add keyed state that uses custom stateful serializer to allround test job
    
    This closes #6909.
---
 .../tests/DataStreamAllroundTestProgram.java       |   8 +-
 .../StatefulComplexPayloadSerializer.java          | 170 +++++++++++++++++++++
 2 files changed, 175 insertions(+), 3 deletions(-)

diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
index 513d0cf..70fdade 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
@@ -29,6 +29,7 @@ import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
 import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.streaming.tests.artificialstate.ComplexPayload;
+import org.apache.flink.streaming.tests.artificialstate.StatefulComplexPayloadSerializer;
 import org.apache.flink.streaming.tests.avro.ComplexPayloadAvro;
 import org.apache.flink.streaming.tests.avro.InnerPayLoadAvro;
 import org.apache.flink.util.Collector;
@@ -91,11 +92,12 @@ public class DataStreamAllroundTestProgram {
 							}
 							return new ComplexPayload(event, KEYED_STATE_OPER_NAME);
 						},
-					Collections.singletonList(
-						new KryoSerializer<>(ComplexPayload.class, env.getConfig())), // custom KryoSerializer
+					Arrays.asList(
+						new KryoSerializer<>(ComplexPayload.class, env.getConfig()), // KryoSerializer
+						new StatefulComplexPayloadSerializer()), // custom stateful serializer
 					Collections.singletonList(ComplexPayload.class) // KryoSerializer via type extraction
 				)
-			).returns(Event.class).name(KEYED_STATE_OPER_NAME + "_Kryo");
+			).returns(Event.class).name(KEYED_STATE_OPER_NAME + "_Kryo_and_Custom_Stateful");
 
 		// add a keyed stateful map operator, which uses Avro for state serialization
 		eventStream = eventStream
diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/artificialstate/StatefulComplexPayloadSerializer.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/artificialstate/StatefulComplexPayloadSerializer.java
new file mode 100644
index 0000000..89b16b6
--- /dev/null
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/artificialstate/StatefulComplexPayloadSerializer.java
@@ -0,0 +1,170 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.streaming.tests.artificialstate;
+
+import org.apache.flink.api.common.typeutils.CompatibilityResult;
+import org.apache.flink.api.common.typeutils.ParameterlessTypeSerializerConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * A custom stateful serializer to test that serializers are not used concurrently.
+ */
+public class StatefulComplexPayloadSerializer extends TypeSerializer<ComplexPayload> {
+
+	private static final long serialVersionUID = 8766687317209282373L;
+
+	/** This holds the thread that currently has exclusive ownership over the serializer. */
+	private final AtomicReference<Thread> currentOwnerThread;
+
+	public StatefulComplexPayloadSerializer() {
+		this.currentOwnerThread = new AtomicReference<>(null);
+	}
+
+	@Override
+	public boolean isImmutableType() {
+		return false;
+	}
+
+	@Override
+	public TypeSerializer<ComplexPayload> duplicate() {
+		return new StatefulComplexPayloadSerializer();
+	}
+
+	@Override
+	public ComplexPayload createInstance() {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public ComplexPayload copy(ComplexPayload from) {
+		try {
+			Thread currentThread = Thread.currentThread();
+			if (currentOwnerThread.compareAndSet(null, currentThread)) {
+				return InstantiationUtil.deserializeObject(
+					InstantiationUtil.serializeObject(from), currentThread.getContextClassLoader());
+			} else {
+				throw new IllegalStateException("Concurrent access to type serializer detected!");
+			}
+		} catch (Exception e) {
+			throw new RuntimeException(e);
+		} finally {
+			currentOwnerThread.set(null);
+		}
+	}
+
+	@Override
+	public ComplexPayload copy(ComplexPayload from, ComplexPayload reuse) {
+		return copy(from);
+	}
+
+	@Override
+	public int getLength() {
+		return -1;
+	}
+
+	@Override
+	public void serialize(ComplexPayload record, DataOutputView target) throws IOException {
+		try {
+			if (currentOwnerThread.compareAndSet(null, Thread.currentThread())) {
+				target.write(InstantiationUtil.serializeObject(record));
+			} else {
+				throw new IllegalStateException("Concurrent access to type serializer detected!");
+			}
+		} finally {
+			currentOwnerThread.set(null);
+		}
+	}
+
+	@Override
+	public ComplexPayload deserialize(DataInputView source) throws IOException {
+		try (final DataInputViewStream inViewWrapper = new DataInputViewStream(source)) {
+			Thread currentThread = Thread.currentThread();
+			if (currentOwnerThread.compareAndSet(null, currentThread)) {
+				return InstantiationUtil.deserializeObject(
+					inViewWrapper,
+					currentThread.getContextClassLoader());
+			} else {
+				throw new IllegalStateException("Concurrent access to type serializer detected!");
+			}
+		} catch (ClassNotFoundException e) {
+			throw new IOException("Could not deserialize object.", e);
+		} finally {
+			currentOwnerThread.set(null);
+		}
+	}
+
+	@Override
+	public ComplexPayload deserialize(ComplexPayload reuse, DataInputView source) throws IOException {
+		return deserialize(source);
+	}
+
+	@Override
+	public void copy(DataInputView source, DataOutputView target) throws IOException {
+		serialize(deserialize(source), target);
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		return obj == this;
+	}
+
+	@Override
+	public boolean canEqual(Object obj) {
+		return getClass().equals(obj.getClass());
+	}
+
+	@Override
+	public int hashCode() {
+		return 42;
+	}
+
+	@Override
+	public TypeSerializerConfigSnapshot<ComplexPayload> snapshotConfiguration() {
+		// type serializer singletons should always be parameter-less
+		return new ParameterlessTypeSerializerConfig<>(getSerializationFormatIdentifier());
+	}
+
+	@Override
+	public CompatibilityResult<ComplexPayload> ensureCompatibility(TypeSerializerConfigSnapshot<?> configSnapshot) {
+		if (configSnapshot instanceof ParameterlessTypeSerializerConfig
+			&& isCompatibleSerializationFormatIdentifier(
+			((ParameterlessTypeSerializerConfig<?>) configSnapshot).getSerializationFormatIdentifier())) {
+
+			return CompatibilityResult.compatible();
+		} else {
+			return CompatibilityResult.requiresMigration();
+		}
+	}
+
+	private boolean isCompatibleSerializationFormatIdentifier(String identifier) {
+		return identifier.equals(getSerializationFormatIdentifier());
+	}
+
+	private String getSerializationFormatIdentifier() {
+		return getClass().getCanonicalName();
+	}
+}