[FLINK-7968] [core] Move DataOutputSerializer and DataInputDeserializ…

…er to 'flink-core' These core flink utils are independent of any other runtime classes and are also used both in flink-runtime and in flink-queryable-state (which duplicated the code).
laubersder · Nov 2, 2017 · 37df826 · 37df826
1 parent 198b74a
commit 37df826
Show file tree

Hide file tree

Showing 38 changed files with 66 additions and 807 deletions.
diff --git a/...apache/flink/streaming/util/serialization/TypeInformationKeyValueSerializationSchema.java b/...apache/flink/streaming/util/serialization/TypeInformationKeyValueSerializationSchema.java
@@ -24,8 +24,8 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.TupleTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.runtime.util.DataInputDeserializer;
-import org.apache.flink.runtime.util.DataOutputSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
 
 import java.io.IOException;
 

diff --git a/.../serialization/DataInputDeserializer.java → ...nk/core/memory/DataInputDeserializer.java b/.../serialization/DataInputDeserializer.java → ...nk/core/memory/DataInputDeserializer.java
@@ -16,10 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.queryablestate.client.state.serialization;
-
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.MemoryUtils;
+package org.apache.flink.core.memory;
 
 import java.io.EOFException;
 import java.io.IOException;
@@ -29,8 +26,6 @@
 
 /**
  * A simple and efficient deserializer for the {@link java.io.DataInput} interface.
- *
- * <p><b>THIS WAS COPIED FROM RUNTIME SO THAT WE AVOID THE DEPENDENCY.</b>
  */
 public class DataInputDeserializer implements DataInputView, java.io.Serializable {
 

diff --git a/...e/serialization/DataOutputSerializer.java → ...ink/core/memory/DataOutputSerializer.java b/...e/serialization/DataOutputSerializer.java → ...ink/core/memory/DataOutputSerializer.java
@@ -16,11 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.queryablestate.client.state.serialization;
-
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.core.memory.MemoryUtils;
+package org.apache.flink.core.memory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,8 +30,6 @@
 
 /**
  * A simple and efficient serializer for the {@link java.io.DataOutput} interface.
- *
- * <p><b>THIS WAS COPIED FROM RUNTIME SO THAT WE AVOID THE DEPENDENCY.</b>
  */
 public class DataOutputSerializer implements DataOutputView {
 
@@ -170,7 +164,6 @@ public void writeChars(String s) throws IOException {
  if (this.position >= this.buffer.length - 2 * sLen) {
  resize(2 * sLen);
  }
-
  for (int i = 0; i < sLen; i++) {
  writeChar(s.charAt(i));
  }
@@ -218,7 +211,7 @@ public void writeShort(int v) throws IOException {
  resize(2);
  }
  this.buffer[this.position++] = (byte) ((v >>> 8) & 0xff);
- this.buffer[this.position++] = (byte) ((v >>> 0) & 0xff);
+ this.buffer[this.position++] = (byte) (v & 0xff);
  }
 
  @Override
@@ -250,9 +243,9 @@ else if (this.position > this.buffer.length - utflen - 2) {
  int count = this.position;
 
  bytearr[count++] = (byte) ((utflen >>> 8) & 0xFF);
- bytearr[count++] = (byte) ((utflen >>> 0) & 0xFF);
+ bytearr[count++] = (byte) (utflen & 0xFF);
 
- int i = 0;
+ int i;
  for (i = 0; i < strlen; i++) {
  c = str.charAt(i);
  if (!((c >= 0x0001) && (c <= 0x007F))) {
@@ -268,11 +261,11 @@ else if (this.position > this.buffer.length - utflen - 2) {
 
  } else if (c > 0x07FF) {
  bytearr[count++] = (byte) (0xE0 | ((c >> 12) & 0x0F));
- bytearr[count++] = (byte) (0x80 | ((c >> 6) & 0x3F));
- bytearr[count++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+ bytearr[count++] = (byte) (0x80 | ((c >>  6) & 0x3F));
+ bytearr[count++] = (byte) (0x80 | (c & 0x3F));
  } else {
  bytearr[count++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
- bytearr[count++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+ bytearr[count++] = (byte) (0x80 | (c & 0x3F));
  }
  }
 

diff --git a/...ntime/util/DataInputDeserializerTest.java → ...ore/memory/DataInputDeserializerTest.java b/...ntime/util/DataInputDeserializerTest.java → ...ore/memory/DataInputDeserializerTest.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.util;
+package org.apache.flink.core.memory;
 
 import org.junit.Assert;
 import org.junit.Test;

diff --git a/...e/util/DataInputOutputSerializerTest.java → ...memory/DataInputOutputSerializerTest.java b/...e/util/DataInputOutputSerializerTest.java → ...memory/DataInputOutputSerializerTest.java
@@ -16,21 +16,22 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.util;
+package org.apache.flink.core.memory;
 
-import org.apache.flink.core.memory.MemorySegmentFactory;
-import org.junit.Assert;
+import org.apache.flink.testutils.serialization.types.SerializationTestType;
+import org.apache.flink.testutils.serialization.types.SerializationTestTypeFactory;
+import org.apache.flink.testutils.serialization.types.Util;
 
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.io.network.api.serialization.types.SerializationTestType;
-import org.apache.flink.runtime.io.network.api.serialization.types.SerializationTestTypeFactory;
-import org.apache.flink.runtime.io.network.api.serialization.types.Util;
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
 
+/**
+ * Tests for the combination of {@link DataOutputSerializer} and {@link DataInputDeserializer}.
+ */
 public class DataInputOutputSerializerTest {
 
  @Test
@@ -88,7 +89,7 @@ public void testWrapAsByteBuffer() {
  @Test
  public void testRandomValuesWriteRead() {
  final int numElements = 100000;
- final ArrayDeque<SerializationTestType> reference = new ArrayDeque<SerializationTestType>();
+ final ArrayDeque<SerializationTestType> reference = new ArrayDeque<>();
 
  DataOutputSerializer serializer = new DataOutputSerializer(1);
 

diff --git a/.../serialization/types/AsciiStringType.java → .../serialization/types/AsciiStringType.java b/.../serialization/types/AsciiStringType.java → .../serialization/types/AsciiStringType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.core.memory.DataInputView;

diff --git a/.../api/serialization/types/BooleanType.java → ...tils/serialization/types/BooleanType.java b/.../api/serialization/types/BooleanType.java → ...tils/serialization/types/BooleanType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...pi/serialization/types/ByteArrayType.java → ...ls/serialization/types/ByteArrayType.java b/...pi/serialization/types/ByteArrayType.java → ...ls/serialization/types/ByteArrayType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Arrays;

diff --git a/...serialization/types/ByteSubArrayType.java → ...serialization/types/ByteSubArrayType.java b/...serialization/types/ByteSubArrayType.java → ...serialization/types/ByteSubArrayType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Arrays;

diff --git a/...ork/api/serialization/types/ByteType.java → ...stutils/serialization/types/ByteType.java b/...ork/api/serialization/types/ByteType.java → ...stutils/serialization/types/ByteType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...ork/api/serialization/types/CharType.java → ...stutils/serialization/types/CharType.java b/...ork/api/serialization/types/CharType.java → ...stutils/serialization/types/CharType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...k/api/serialization/types/DoubleType.java → ...utils/serialization/types/DoubleType.java b/...k/api/serialization/types/DoubleType.java → ...utils/serialization/types/DoubleType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...rk/api/serialization/types/FloatType.java → ...tutils/serialization/types/FloatType.java b/...rk/api/serialization/types/FloatType.java → ...tutils/serialization/types/FloatType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...work/api/serialization/types/IntType.java → ...estutils/serialization/types/IntType.java b/...work/api/serialization/types/IntType.java → ...estutils/serialization/types/IntType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...ork/api/serialization/types/LongType.java → ...stutils/serialization/types/LongType.java b/...ork/api/serialization/types/LongType.java → ...stutils/serialization/types/LongType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...lization/types/SerializationTestType.java → ...lization/types/SerializationTestType.java b/...lization/types/SerializationTestType.java → ...lization/types/SerializationTestType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.util.Random;
 

diff --git a/...n/types/SerializationTestTypeFactory.java → ...n/types/SerializationTestTypeFactory.java b/...n/types/SerializationTestTypeFactory.java → ...n/types/SerializationTestTypeFactory.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 public enum SerializationTestTypeFactory {
 

diff --git a/...rk/api/serialization/types/ShortType.java → ...tutils/serialization/types/ShortType.java b/...rk/api/serialization/types/ShortType.java → ...tutils/serialization/types/ShortType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...serialization/types/UnsignedByteType.java → ...serialization/types/UnsignedByteType.java b/...serialization/types/UnsignedByteType.java → ...serialization/types/UnsignedByteType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...erialization/types/UnsignedShortType.java → ...erialization/types/UnsignedShortType.java b/...erialization/types/UnsignedShortType.java → ...erialization/types/UnsignedShortType.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.io.IOException;
 import java.util.Random;

diff --git a/...network/api/serialization/types/Util.java → ...k/testutils/serialization/types/Util.java b/...network/api/serialization/types/Util.java → ...k/testutils/serialization/types/Util.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.runtime.io.network.api.serialization.types;
+package org.apache.flink.testutils.serialization.types;
 
 import java.util.Iterator;
 import java.util.NoSuchElementException;

diff --git a/...in/java/org/apache/flink/queryablestate/client/state/serialization/KvStateSerializer.java b/...in/java/org/apache/flink/queryablestate/client/state/serialization/KvStateSerializer.java
@@ -20,6 +20,8 @@
 
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;

diff --git a/...apache/flink/runtime/io/network/api/serialization/AdaptiveSpanningRecordDeserializer.java b/...apache/flink/runtime/io/network/api/serialization/AdaptiveSpanningRecordDeserializer.java
@@ -22,8 +22,8 @@
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.runtime.io.network.buffer.Buffer;
-import org.apache.flink.runtime.util.DataInputDeserializer;
-import org.apache.flink.runtime.util.DataOutputSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
 
 import java.io.EOFException;
 import java.io.IOException;

diff --git a/.../src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java b/.../src/main/java/org/apache/flink/runtime/io/network/api/serialization/EventSerializer.java
@@ -30,8 +30,8 @@
 import org.apache.flink.runtime.io.network.api.EndOfSuperstepEvent;
 import org.apache.flink.runtime.io.network.buffer.Buffer;
 import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
-import org.apache.flink.runtime.util.DataInputDeserializer;
-import org.apache.flink.runtime.util.DataOutputSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
 import org.apache.flink.util.InstantiationUtil;
 
 import java.io.IOException;

diff --git a/.../java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializer.java b/.../java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializer.java
@@ -26,7 +26,7 @@
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
 import org.apache.flink.runtime.io.network.buffer.Buffer;
-import org.apache.flink.runtime.util.DataOutputSerializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
 
 /**
  * Record serializer which serializes the complete record to an intermediate

diff --git a/...link/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java b/...link/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
@@ -23,7 +23,7 @@
 import org.apache.flink.core.memory.DataInputViewStreamWrapper;
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.runtime.io.network.buffer.Buffer;
-import org.apache.flink.runtime.util.DataInputDeserializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
 import org.apache.flink.util.StringUtils;
 
 import java.io.BufferedInputStream;

diff --git a/...-runtime/src/main/java/org/apache/flink/runtime/metrics/dump/MetricDumpSerialization.java b/...-runtime/src/main/java/org/apache/flink/runtime/metrics/dump/MetricDumpSerialization.java
@@ -19,15 +19,15 @@
 package org.apache.flink.runtime.metrics.dump;
 
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.DataInputDeserializer;
 import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Gauge;
 import org.apache.flink.metrics.Histogram;
 import org.apache.flink.metrics.HistogramStatistics;
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.Metric;
-import org.apache.flink.runtime.util.DataInputDeserializer;
-import org.apache.flink.runtime.util.DataOutputSerializer;
 import org.apache.flink.util.Preconditions;
 
 import org.slf4j.Logger;