This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 63bdbc9d4a2b2c025e450cd1cf6e6008940389de
Author: Dawid Wysakowicz <dwysakow...@apache.org>
AuthorDate: Mon Sep 21 15:05:35 2020 +0200
[FLINK-19472] Implement a one input sorting DataInput
I implement a SortingDataInput which can wrap a regular network input
and sort the incoming records before passing them on to the DataOutput.
The sorting input performs the final sorting when it receives an
InputStatus.END_OF_INPUT from the chained input.
The sorter uses a binary comparison of serialized keys of the incoming
records. It uses first n bytes of the serialized key as a normalized
key.
Watermarks, stream statuses, or latency markers are not propagated, only
the largest seen watermark is
emitted after all records.
This closes #13521
---
.../api/common/typeutils/ComparatorTestBase.java | 66 +++--
.../operators/sort/NormalizedKeySorter.java | 2 +-
.../runtime/operators/sort/SpillingThread.java | 10 +-
.../operators/sort/BytesKeyNormalizationUtil.java | 85 +++++++
.../sort/FixedLengthByteKeyComparator.java | 178 ++++++++++++++
.../api/operators/sort/KeyAndValueSerializer.java | 188 +++++++++++++++
.../api/operators/sort/SortingDataInput.java | 229 ++++++++++++++++++
.../sort/VariableLengthByteKeyComparator.java | 182 ++++++++++++++
.../sort/FixedLengthByteKeyComparatorTest.java | 70 ++++++
.../sort/FixedLengthKeyAndValueSerializerTest.java | 73 ++++++
.../sort/SerializerComparatorTestData.java | 109 +++++++++
.../api/operators/sort/SortingDataInputITCase.java | 243 +++++++++++++++++++
.../api/operators/sort/SortingDataInputTest.java | 268 +++++++++++++++++++++
.../sort/VariableLengthByteKeyComparatorTest.java | 68 ++++++
.../VariableLengthKeyAndValueSerializerTest.java | 70 ++++++
15 files changed, 1812 insertions(+), 29 deletions(-)
diff --git
a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java
b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java
index 279c2b3..fd3581c 100644
---
a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java
+++
b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/ComparatorTestBase.java
@@ -18,14 +18,7 @@
package org.apache.flink.api.common.typeutils;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-import static org.junit.Assert.*;
-
+import org.apache.flink.api.common.operators.Order;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.MemorySegment;
@@ -35,6 +28,17 @@ import org.apache.flink.util.TestLogger;
import org.junit.Assert;
import org.junit.Test;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
/**
* Abstract test base for comparators.
*
@@ -45,8 +49,8 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
// Same as in the NormalizedKeySorter
private static final int DEFAULT_MAX_NORMALIZED_KEY_LEN = 8;
- protected boolean[] getTestedOrder() {
- return new boolean[] {true, false};
+ protected Order[] getTestedOrder() {
+ return new Order[] {Order.ASCENDING, Order.DESCENDING};
}
protected abstract TypeComparator<T> createComparator(boolean
ascending);
@@ -67,7 +71,8 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
@Test
public void testDuplicate() {
try {
- TypeComparator<T> comparator = getComparator(true);
+ boolean ascending = isAscending(getTestedOrder()[0]);
+ TypeComparator<T> comparator = getComparator(ascending);
TypeComparator<T> clone = comparator.duplicate();
T[] data = getSortedData();
@@ -88,9 +93,9 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
@Test
public void testEquality() {
- for (boolean ascending : getTestedOrder()) {
- testEquals(true);
- testEquals(false);
+ for (Order order : getTestedOrder()) {
+ boolean ascending = isAscending(order);
+ testEquals(ascending);
}
}
@@ -128,8 +133,9 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
public void testEqualityWithReference() {
try {
TypeSerializer<T> serializer = createSerializer();
- TypeComparator<T> comparator = getComparator(true);
- TypeComparator<T> comparator2 = getComparator(true);
+ boolean ascending = isAscending(getTestedOrder()[0]);
+ TypeComparator<T> comparator = getComparator(ascending);
+ TypeComparator<T> comparator2 =
getComparator(ascending);
T[] data = getSortedData();
for (T d : data) {
comparator.setReference(d);
@@ -151,7 +157,8 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
// --------------------------------- inequality tests
----------------------------------------
@Test
public void testInequality() {
- for (boolean ascending : getTestedOrder()) {
+ for (Order order : getTestedOrder()) {
+ boolean ascending = isAscending(order);
testGreatSmallAscDesc(ascending, true);
testGreatSmallAscDesc(ascending, false);
}
@@ -202,7 +209,8 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
@Test
public void testInequalityWithReference() {
- for (boolean ascending : getTestedOrder()) {
+ for (Order order : getTestedOrder()) {
+ boolean ascending = isAscending(order);
testGreatSmallAscDescWithReference(ascending, true);
testGreatSmallAscDescWithReference(ascending, false);
}
@@ -276,7 +284,8 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
@Test
public void testNormalizedKeysEqualsFullLength() {
// Ascending or descending does not matter in this case
- TypeComparator<T> comparator = getComparator(true);
+ boolean ascending = isAscending(getTestedOrder()[0]);
+ TypeComparator<T> comparator = getComparator(ascending);
if (!comparator.supportsNormalizedKey()) {
return;
}
@@ -285,7 +294,8 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
@Test
public void testNormalizedKeysEqualsHalfLength() {
- TypeComparator<T> comparator = getComparator(true);
+ boolean ascending = isAscending(getTestedOrder()[0]);
+ TypeComparator<T> comparator = getComparator(ascending);
if (!comparator.supportsNormalizedKey()) {
return;
}
@@ -294,7 +304,8 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
public void testNormalizedKeysEquals(boolean halfLength) {
try {
- TypeComparator<T> comparator = getComparator(true);
+ boolean ascending = isAscending(getTestedOrder()[0]);
+ TypeComparator<T> comparator = getComparator(ascending);
T[] data = getSortedData();
int normKeyLen = getNormKeyLen(halfLength, data,
comparator);
@@ -315,7 +326,7 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
@Test
public void testNormalizedKeysGreatSmallFullLength() {
// ascending/descending in comparator doesn't matter for
normalized keys
- boolean ascending = getTestedOrder()[0];
+ boolean ascending = isAscending(getTestedOrder()[0]);
TypeComparator<T> comparator = getComparator(ascending);
if (!comparator.supportsNormalizedKey()) {
return;
@@ -327,7 +338,7 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
@Test
public void testNormalizedKeysGreatSmallAscDescHalfLength() {
// ascending/descending in comparator doesn't matter for
normalized keys
- boolean ascending = getTestedOrder()[0];
+ boolean ascending = isAscending(getTestedOrder()[0]);
TypeComparator<T> comparator = getComparator(ascending);
if (!comparator.supportsNormalizedKey()) {
return;
@@ -382,7 +393,7 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
T[] data = getSortedData();
T reuse = getSortedData()[0];
- boolean ascending = getTestedOrder()[0];
+ boolean ascending = isAscending(getTestedOrder()[0]);
TypeComparator<T> comp1 = getComparator(ascending);
if(!comp1.supportsSerializationWithKeyNormalization()){
return;
@@ -413,7 +424,7 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
@Test
@SuppressWarnings("unchecked")
public void testKeyExtraction() {
- boolean ascending = getTestedOrder()[0];
+ boolean ascending = isAscending(getTestedOrder()[0]);
TypeComparator<T> comparator = getComparator(ascending);
T[] data = getSortedData();
@@ -491,6 +502,11 @@ public abstract class ComparatorTestBase<T> extends
TestLogger {
}
//
--------------------------------------------------------------------------------------------
+
+ private static boolean isAscending(Order order) {
+ return order == Order.ASCENDING;
+ }
+
public static final class TestOutputView extends DataOutputStream
implements DataOutputView {
public TestOutputView() {
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NormalizedKeySorter.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NormalizedKeySorter.java
index 512e73e..8043521 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NormalizedKeySorter.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/NormalizedKeySorter.java
@@ -531,7 +531,7 @@ public final class NormalizedKeySorter<T> implements
InMemorySorter<T> {
}
}
}
-
+
/**
* Writes a subset of the records in this buffer in their logical order
to the given output.
*
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
index 0c52fae..3f2d059 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/SpillingThread.java
@@ -310,9 +310,13 @@ final class SpillingThread<E> extends ThreadBase<E> {
disposeSortBuffers(true);
// set lazy iterator
- this.dispatcher.sendResult(iterators.isEmpty() ?
EmptyMutableObjectIterator.get() :
- iterators.size() == 1 ? iterators.get(0) :
- new MergeIterator<>(iterators,
this.comparator));
+ if (iterators.isEmpty()) {
+
this.dispatcher.sendResult(EmptyMutableObjectIterator.get());
+ } else if (iterators.size() == 1) {
+ this.dispatcher.sendResult(iterators.get(0));
+ } else {
+ this.dispatcher.sendResult(new
MergeIterator<>(iterators, this.comparator));
+ }
}
private List<ChannelWithBlockCount>
startSpilling(Queue<CircularElement<E>> cache) throws IOException,
InterruptedException {
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/BytesKeyNormalizationUtil.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/BytesKeyNormalizationUtil.java
new file mode 100644
index 0000000..ef0acf7
--- /dev/null
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/BytesKeyNormalizationUtil.java
@@ -0,0 +1,85 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import static
org.apache.flink.streaming.api.operators.sort.FixedLengthByteKeyComparator.TIMESTAMP_BYTE_SIZE;
+
+/**
+ * Utility class for common key normalization used both in {@link
VariableLengthByteKeyComparator}
+ * and {@link FixedLengthByteKeyComparator}.
+ */
+final class BytesKeyNormalizationUtil {
+ /**
+ * Writes the normalized key of given record. The normalized key
consists of the key serialized as bytes and
+ * the timestamp of the record.
+ *
+ * <p>NOTE: The key does not represent a logical order. It can be used
only for grouping keys!
+ */
+ static <IN> void putNormalizedKey(
+ Tuple2<byte[], StreamRecord<IN>> record,
+ int dataLength,
+ MemorySegment target,
+ int offset,
+ int numBytes) {
+ byte[] data = record.f0;
+
+ if (dataLength >= numBytes) {
+ putBytesArray(target, offset, numBytes, data);
+ } else {
+ // whole key fits into the normalized key
+ putBytesArray(target, offset, dataLength, data);
+ int lastOffset = offset + numBytes;
+ offset += dataLength;
+ long valueOfTimestamp =
record.f1.asRecord().getTimestamp() - Long.MIN_VALUE;
+ if (dataLength + TIMESTAMP_BYTE_SIZE <= numBytes) {
+ // whole timestamp fits into the normalized key
+ target.putLong(offset, valueOfTimestamp);
+ offset += TIMESTAMP_BYTE_SIZE;
+ // fill in the remaining space with zeros
+ while (offset < lastOffset) {
+ target.put(offset++, (byte) 0);
+ }
+ } else {
+ // only part of the timestamp fits into
normalized key
+ for (int i = 0; offset < lastOffset; offset++,
i++) {
+ target.put(offset, (byte)
(valueOfTimestamp >>> ((7 - i) << 3)));
+ }
+ }
+ }
+ }
+
+ private static void putBytesArray(MemorySegment target, int offset, int
numBytes, byte[] data) {
+ for (int i = 0; i < numBytes; i++) {
+ // We're converting the signed byte in data into an
unsigned representation.
+ // A Java byte goes from -127 to 128, i.e. is signed.
By subtracting -127 (MIN_VALUE)
+ // here we're shifting the number to be from 0 to 255.
The normalized key sorter sorts
+ // bytes as "unsigned", so we need to convert here to
maintain a correct ordering.
+ int highByte = data[i] & 0xff;
+ highByte -= Byte.MIN_VALUE;
+ target.put(offset + i, (byte) highByte);
+ }
+ }
+
+ private BytesKeyNormalizationUtil() {
+ }
+}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/FixedLengthByteKeyComparator.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/FixedLengthByteKeyComparator.java
new file mode 100644
index 0000000..6b56c24
--- /dev/null
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/FixedLengthByteKeyComparator.java
@@ -0,0 +1,178 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * A comparator used in {@link SortingDataInput} which compares records keys
and timestamps.
+ * It uses binary format produced by the {@link KeyAndValueSerializer}.
+ *
+ * <p>It assumes keys are always of a fixed length and thus the length of the
record is not serialized.
+ */
+final class FixedLengthByteKeyComparator<IN> extends
TypeComparator<Tuple2<byte[], StreamRecord<IN>>> {
+ static final int TIMESTAMP_BYTE_SIZE = 8;
+ private final int keyLength;
+ private byte[] keyReference;
+ private long timestampReference;
+
+ FixedLengthByteKeyComparator(int keyLength) {
+ this.keyLength = keyLength;
+ }
+
+ @Override
+ public int hash(Tuple2<byte[], StreamRecord<IN>> record) {
+ return record.hashCode();
+ }
+
+ @Override
+ public void setReference(Tuple2<byte[], StreamRecord<IN>> toCompare) {
+ this.keyReference = toCompare.f0;
+ this.timestampReference =
toCompare.f1.asRecord().getTimestamp();
+ }
+
+ @Override
+ public boolean equalToReference(Tuple2<byte[], StreamRecord<IN>>
candidate) {
+ return Arrays.equals(keyReference, candidate.f0) &&
+ timestampReference ==
candidate.f1.asRecord().getTimestamp();
+ }
+
+ @Override
+ public int compareToReference(TypeComparator<Tuple2<byte[],
StreamRecord<IN>>> referencedComparator) {
+ byte[] otherKey = ((FixedLengthByteKeyComparator<IN>)
referencedComparator).keyReference;
+ long otherTimestamp = ((FixedLengthByteKeyComparator<IN>)
referencedComparator).timestampReference;
+
+ int keyCmp = compare(otherKey, this.keyReference);
+ if (keyCmp != 0) {
+ return keyCmp;
+ }
+ return Long.compare(otherTimestamp, this.timestampReference);
+ }
+
+ @Override
+ public int compare(
+ Tuple2<byte[], StreamRecord<IN>> first,
+ Tuple2<byte[], StreamRecord<IN>> second) {
+ int keyCmp = compare(first.f0, second.f0);
+ if (keyCmp != 0) {
+ return keyCmp;
+ }
+ return Long.compare(first.f1.asRecord().getTimestamp(),
second.f1.asRecord().getTimestamp());
+ }
+
+ private int compare(byte[] first, byte[] second) {
+ for (int i = 0; i < keyLength; i++) {
+ int cmp = Byte.compare(first[i], second[i]);
+
+ if (cmp != 0) {
+ return cmp < 0 ? -1 : 1;
+ }
+ }
+
+ return 0;
+ }
+
+ @Override
+ public int compareSerialized(DataInputView firstSource, DataInputView
secondSource) throws IOException {
+ int minCount = keyLength;
+ while (minCount-- > 0) {
+ byte firstValue = firstSource.readByte();
+ byte secondValue = secondSource.readByte();
+
+ int cmp = Byte.compare(firstValue, secondValue);
+ if (cmp != 0) {
+ return cmp < 0 ? -1 : 1;
+ }
+ }
+
+ return Long.compare(firstSource.readLong(),
secondSource.readLong());
+ }
+
+ @Override
+ public boolean supportsNormalizedKey() {
+ return true;
+ }
+
+ @Override
+ public int getNormalizeKeyLen() {
+ return keyLength + TIMESTAMP_BYTE_SIZE;
+ }
+
+ @Override
+ public boolean isNormalizedKeyPrefixOnly(int keyBytes) {
+ return keyBytes < getNormalizeKeyLen();
+ }
+
+ @Override
+ public void putNormalizedKey(Tuple2<byte[], StreamRecord<IN>> record,
MemorySegment target, int offset, int numBytes) {
+ BytesKeyNormalizationUtil.putNormalizedKey(record, keyLength,
target, offset, numBytes);
+ }
+
+ @Override
+ public boolean invertNormalizedKey() {
+ return false;
+ }
+
+ @Override
+ public TypeComparator<Tuple2<byte[], StreamRecord<IN>>> duplicate() {
+ return new FixedLengthByteKeyComparator<>(this.keyLength);
+ }
+
+ @Override
+ public int extractKeys(Object record, Object[] target, int index) {
+ target[index] = record;
+ return 1;
+ }
+
+ @Override
+ public TypeComparator<?>[] getFlatComparators() {
+ return new TypeComparator[] {this};
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // unsupported normalization
+ //
--------------------------------------------------------------------------------------------
+
+ @Override
+ public boolean supportsSerializationWithKeyNormalization() {
+ return false;
+ }
+
+ @Override
+ public void writeWithKeyNormalization(
+ Tuple2<byte[], StreamRecord<IN>> record,
+ DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Tuple2<byte[], StreamRecord<IN>> readWithKeyDenormalization(
+ Tuple2<byte[], StreamRecord<IN>> reuse,
+ DataInputView source) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/KeyAndValueSerializer.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/KeyAndValueSerializer.java
new file mode 100644
index 0000000..311da3e
--- /dev/null
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/KeyAndValueSerializer.java
@@ -0,0 +1,188 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * A serializer used in {@link SortingDataInput} for serializing elements
alongside their key and
+ * timestamp. It serializes the record in a format known by the {@link
FixedLengthByteKeyComparator}
+ * and {@link VariableLengthByteKeyComparator}.
+ *
+ * <p>If the key is of known constant length, the length is not serialized
with the data.
+ * Therefore the serialized data is as follows:
+ *
+ * <pre>
+ * [key-length] | <key> | <timestamp> | <record>
+ * </pre>
+ */
+final class KeyAndValueSerializer<IN> extends TypeSerializer<Tuple2<byte[],
StreamRecord<IN>>> {
+ private static final int TIMESTAMP_LENGTH = 8;
+ private final TypeSerializer<IN> valueSerializer;
+
+ // This represents either a variable length (-1) or a fixed one (>= 0).
+ private final int serializedKeyLength;
+
+ KeyAndValueSerializer(TypeSerializer<IN> valueSerializer, int
serializedKeyLength) {
+ this.valueSerializer = valueSerializer;
+ this.serializedKeyLength = serializedKeyLength;
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<Tuple2<byte[], StreamRecord<IN>>> duplicate() {
+ return new KeyAndValueSerializer<>(valueSerializer.duplicate(),
this.serializedKeyLength);
+ }
+
+ @Override
+ public Tuple2<byte[], StreamRecord<IN>> copy(Tuple2<byte[],
StreamRecord<IN>> from) {
+ StreamRecord<IN> fromRecord = from.f1;
+ return Tuple2.of(
+ Arrays.copyOf(from.f0, from.f0.length),
+
fromRecord.copy(valueSerializer.copy(fromRecord.getValue()))
+ );
+ }
+
+ @Override
+ public Tuple2<byte[], StreamRecord<IN>> createInstance() {
+ return Tuple2.of(new byte[0], new
StreamRecord<>(valueSerializer.createInstance()));
+ }
+
+ @Override
+ public Tuple2<byte[], StreamRecord<IN>> copy(
+ Tuple2<byte[], StreamRecord<IN>> from,
+ Tuple2<byte[], StreamRecord<IN>> reuse) {
+ StreamRecord<IN> fromRecord = from.f1;
+ StreamRecord<IN> reuseRecord = reuse.f1;
+
+ IN valueCopy = valueSerializer.copy(fromRecord.getValue(),
reuseRecord.getValue());
+ fromRecord.copyTo(valueCopy, reuseRecord);
+ reuse.f0 = Arrays.copyOf(from.f0, from.f0.length);
+ reuse.f1 = reuseRecord;
+ return reuse;
+ }
+
+ @Override
+ public int getLength() {
+ if (valueSerializer.getLength() < 0 || serializedKeyLength < 0)
{
+ return -1;
+ }
+ return valueSerializer.getLength() + serializedKeyLength +
TIMESTAMP_LENGTH;
+ }
+
+ @Override
+ public void serialize(Tuple2<byte[], StreamRecord<IN>> record,
DataOutputView target) throws IOException {
+ if (serializedKeyLength < 0) {
+ target.writeInt(record.f0.length);
+ }
+ target.write(record.f0);
+ StreamRecord<IN> toSerialize = record.f1;
+ target.writeLong(toSerialize.getTimestamp());
+ valueSerializer.serialize(toSerialize.getValue(), target);
+ }
+
+ @Override
+ public Tuple2<byte[], StreamRecord<IN>> deserialize(DataInputView
source) throws IOException {
+ final int length = getKeyLength(source);
+ byte[] bytes = new byte[length];
+ source.read(bytes);
+ long timestamp = source.readLong();
+ IN value = valueSerializer.deserialize(source);
+ return Tuple2.of(
+ bytes,
+ new StreamRecord<>(value, timestamp)
+ );
+ }
+
+ @Override
+ public Tuple2<byte[], StreamRecord<IN>> deserialize(Tuple2<byte[],
StreamRecord<IN>> reuse, DataInputView source) throws IOException {
+ final int length = getKeyLength(source);
+ byte[] bytes = new byte[length];
+ source.read(bytes);
+ long timestamp = source.readLong();
+ IN value = valueSerializer.deserialize(source);
+ StreamRecord<IN> reuseRecord = reuse.f1;
+ reuseRecord.replace(value, timestamp);
+ reuse.f0 = bytes;
+ reuse.f1 = reuseRecord;
+ return reuse;
+ }
+
+ private int getKeyLength(DataInputView source) throws IOException {
+ final int length;
+ if (serializedKeyLength < 0) {
+ length = source.readInt();
+ } else {
+ length = serializedKeyLength;
+ }
+ return length;
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws
IOException {
+ final int length;
+ if (serializedKeyLength < 0) {
+ length = source.readInt();
+ target.writeInt(length);
+ } else {
+ length = serializedKeyLength;
+ }
+ for (int i = 0; i < length; i++) {
+ target.writeByte(source.readByte());
+ }
+ target.writeLong(source.readLong());
+ valueSerializer.copy(source, target);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ KeyAndValueSerializer<?> that = (KeyAndValueSerializer<?>) o;
+ return Objects.equals(valueSerializer, that.valueSerializer);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(valueSerializer);
+ }
+
+ @Override
+ public TypeSerializerSnapshot<Tuple2<byte[], StreamRecord<IN>>>
snapshotConfiguration() {
+ throw new UnsupportedOperationException(
+ "The KeyAndValueSerializer should not be used for
persisting into State!");
+ }
+}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingDataInput.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingDataInput.java
new file mode 100644
index 0000000..b59a404
--- /dev/null
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingDataInput.java
@@ -0,0 +1,229 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.sort.ExternalSorter;
+import org.apache.flink.runtime.operators.sort.PushSorter;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.streamstatus.StreamStatus;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MutableObjectIterator;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A {@link StreamTaskInput} which sorts in the incoming records from a
chained input. It postpones
+ * emitting the records until it receives {@link InputStatus#END_OF_INPUT}
from the chained input.
+ * After it is done it emits a single record at a time from the sorter.
+ *
+ * <p>The sorter uses binary comparison of keys, which are extracted and
serialized when received
+ * from the chained input. Moreover the timestamps of incoming records are
used for secondary ordering.
+ * For the comparison it uses either {@link FixedLengthByteKeyComparator} if
the length of the
+ * serialized key is constant, or {@link VariableLengthByteKeyComparator}
otherwise.
+ *
+ * <p>Watermarks, stream statuses, nor latency markers are not propagated
downstream as they do not make
+ * sense with buffered records. The input emits a MAX_WATERMARK after all
records.
+ *
+ * @param <T> The type of the value in incoming {@link StreamRecord
StreamRecords}.
+ * @param <K> The type of the key.
+ */
+public final class SortingDataInput<T, K> implements StreamTaskInput<T> {
+
+ private final StreamTaskInput<T> wrappedInput;
+ private final PushSorter<Tuple2<byte[], StreamRecord<T>>> sorter;
+ private final KeySelector<T, K> keySelector;
+ private final TypeSerializer<K> keySerializer;
+ private final DataOutputSerializer dataOutputSerializer;
+ private final ForwardingDataOutput forwardingDataOutput;
+ private MutableObjectIterator<Tuple2<byte[], StreamRecord<T>>>
sortedInput = null;
+ private boolean emittedLast;
+ private long watermarkSeen = Long.MIN_VALUE;
+
+ public SortingDataInput(
+ StreamTaskInput<T> wrappedInput,
+ TypeSerializer<T> typeSerializer,
+ TypeSerializer<K> keySerializer,
+ KeySelector<T, K> keySelector,
+ MemoryManager memoryManager,
+ IOManager ioManager,
+ boolean objectReuse,
+ double managedMemoryFraction,
+ Configuration jobConfiguration,
+ AbstractInvokable containingTask) {
+ try {
+ this.forwardingDataOutput = new ForwardingDataOutput();
+ this.keySelector = keySelector;
+ this.keySerializer = keySerializer;
+ int keyLength = keySerializer.getLength();
+ final TypeComparator<Tuple2<byte[], StreamRecord<T>>>
comparator;
+ if (keyLength > 0) {
+ this.dataOutputSerializer = new
DataOutputSerializer(keyLength);
+ comparator = new
FixedLengthByteKeyComparator<>(keyLength);
+ } else {
+ this.dataOutputSerializer = new
DataOutputSerializer(64);
+ comparator = new
VariableLengthByteKeyComparator<>();
+ }
+ KeyAndValueSerializer<T> keyAndValueSerializer = new
KeyAndValueSerializer<>(typeSerializer, keyLength);
+ this.wrappedInput = wrappedInput;
+ this.sorter = ExternalSorter.newBuilder(
+ memoryManager,
+ containingTask,
+ keyAndValueSerializer,
+ comparator)
+ .memoryFraction(managedMemoryFraction)
+ .enableSpilling(
+ ioManager,
+
jobConfiguration.get(AlgorithmOptions.SORT_SPILLING_THRESHOLD))
+
.maxNumFileHandles(jobConfiguration.get(AlgorithmOptions.SPILLING_MAX_FAN))
+ .objectReuse(objectReuse)
+ .largeRecords(true)
+ .build();
+ } catch (MemoryAllocationException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public int getInputIndex() {
+ return wrappedInput.getInputIndex();
+ }
+
+ @Override
+ public CompletableFuture<Void> prepareSnapshot(
+ ChannelStateWriter channelStateWriter,
+ long checkpointId) {
+ throw new UnsupportedOperationException("Checkpoints are not
supported for sorting inputs");
+ }
+
+ @Override
+ public void close() throws IOException {
+ IOException ex = null;
+ try {
+ wrappedInput.close();
+ } catch (IOException e) {
+ ex = ExceptionUtils.firstOrSuppressed(e, ex);
+ }
+
+ try {
+ sorter.close();
+ } catch (IOException e) {
+ ex = ExceptionUtils.firstOrSuppressed(e, ex);
+ }
+
+ if (ex != null) {
+ throw ex;
+ }
+ }
+
+ private class ForwardingDataOutput implements DataOutput<T> {
+ @Override
+ public void emitRecord(StreamRecord<T> streamRecord) throws
Exception {
+ K key = keySelector.getKey(streamRecord.getValue());
+
+ keySerializer.serialize(key, dataOutputSerializer);
+ byte[] serializedKey =
dataOutputSerializer.getCopyOfBuffer();
+ dataOutputSerializer.clear();
+
+ sorter.writeRecord(Tuple2.of(serializedKey,
streamRecord));
+ }
+
+ @Override
+ public void emitWatermark(Watermark watermark) {
+ watermarkSeen = Math.max(watermarkSeen,
watermark.getTimestamp());
+ }
+
+ @Override
+ public void emitStreamStatus(StreamStatus streamStatus) {
+
+ }
+
+ @Override
+ public void emitLatencyMarker(LatencyMarker latencyMarker) {
+
+ }
+ }
+
+ @Override
+ public InputStatus emitNext(DataOutput<T> output) throws Exception {
+ if (sortedInput != null) {
+ return emitNextSortedRecord(output);
+ }
+
+ InputStatus inputStatus =
wrappedInput.emitNext(forwardingDataOutput);
+ if (inputStatus == InputStatus.END_OF_INPUT) {
+ endSorting();
+ return emitNextSortedRecord(output);
+ }
+
+ return inputStatus;
+ }
+
+ @Nonnull
+ private InputStatus emitNextSortedRecord(DataOutput<T> output) throws
Exception {
+ if (emittedLast) {
+ return InputStatus.END_OF_INPUT;
+ }
+
+ Tuple2<byte[], StreamRecord<T>> next = sortedInput.next();
+ if (next != null) {
+ output.emitRecord(next.f1);
+ return InputStatus.MORE_AVAILABLE;
+ } else {
+ emittedLast = true;
+ if (watermarkSeen > Long.MIN_VALUE) {
+ output.emitWatermark(new
Watermark(watermarkSeen));
+ }
+ return InputStatus.END_OF_INPUT;
+ }
+ }
+
+ private void endSorting() throws Exception {
+ this.sorter.finishReading();
+ this.sortedInput = sorter.getIterator();
+ }
+
+ @Override
+ public CompletableFuture<?> getAvailableFuture() {
+ if (sortedInput != null) {
+ return AvailabilityProvider.AVAILABLE;
+ } else {
+ return wrappedInput.getAvailableFuture();
+ }
+ }
+}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/VariableLengthByteKeyComparator.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/VariableLengthByteKeyComparator.java
new file mode 100644
index 0000000..9b60506
--- /dev/null
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/VariableLengthByteKeyComparator.java
@@ -0,0 +1,182 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * A comparator used in {@link SortingDataInput} which compares records keys
and timestamps,.
+ * It uses binary format produced by the {@link KeyAndValueSerializer}.
+ *
+ * <p>It assumes keys are of a variable length and thus expects the length of
the record to be serialized.
+ */
+final class VariableLengthByteKeyComparator<IN> extends
TypeComparator<Tuple2<byte[], StreamRecord<IN>>> {
+ private byte[] keyReference;
+ private long timestampReference;
+
+ @Override
+ public int hash(Tuple2<byte[], StreamRecord<IN>> record) {
+ return record.hashCode();
+ }
+
+ @Override
+ public void setReference(Tuple2<byte[], StreamRecord<IN>> toCompare) {
+ this.keyReference = Arrays.copyOf(toCompare.f0,
toCompare.f0.length);
+ this.timestampReference =
toCompare.f1.asRecord().getTimestamp();
+ }
+
+ @Override
+ public boolean equalToReference(Tuple2<byte[], StreamRecord<IN>>
candidate) {
+ return Arrays.equals(keyReference, candidate.f0) &&
+ timestampReference ==
candidate.f1.asRecord().getTimestamp();
+ }
+
+ @Override
+ public int compareToReference(TypeComparator<Tuple2<byte[],
StreamRecord<IN>>> referencedComparator) {
+ byte[] otherKey = ((VariableLengthByteKeyComparator<IN>)
referencedComparator).keyReference;
+ long otherTimestamp = ((VariableLengthByteKeyComparator<IN>)
referencedComparator).timestampReference;
+
+ int keyCmp = compare(otherKey, this.keyReference);
+ if (keyCmp != 0) {
+ return keyCmp;
+ }
+ return Long.compare(otherTimestamp, this.timestampReference);
+ }
+
+ @Override
+ public int compare(
+ Tuple2<byte[], StreamRecord<IN>> first,
+ Tuple2<byte[], StreamRecord<IN>> second) {
+ int keyCmp = compare(first.f0, second.f0);
+ if (keyCmp != 0) {
+ return keyCmp;
+ }
+ return Long.compare(first.f1.asRecord().getTimestamp(),
second.f1.asRecord().getTimestamp());
+ }
+
+ private int compare(byte[] first, byte[] second) {
+ int firstLength = first.length;
+ int secondLength = second.length;
+ int minLength = Math.min(firstLength, secondLength);
+ for (int i = 0; i < minLength; i++) {
+ int cmp = Byte.compare(first[i], second[i]);
+
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+
+ return Integer.compare(firstLength, secondLength);
+ }
+
+ @Override
+ public int compareSerialized(DataInputView firstSource, DataInputView
secondSource) throws IOException {
+ int firstLength = firstSource.readInt();
+ int secondLength = secondSource.readInt();
+ int minLength = Math.min(firstLength, secondLength);
+ while (minLength-- > 0) {
+ byte firstValue = firstSource.readByte();
+ byte secondValue = secondSource.readByte();
+
+ int cmp = Byte.compare(firstValue, secondValue);
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+
+ int lengthCompare = Integer.compare(firstLength, secondLength);
+ if (lengthCompare != 0) {
+ return lengthCompare;
+ } else {
+ return Long.compare(firstSource.readLong(),
secondSource.readLong());
+ }
+ }
+
+ @Override
+ public boolean supportsNormalizedKey() {
+ return true;
+ }
+
+ @Override
+ public int getNormalizeKeyLen() {
+ return Integer.MAX_VALUE;
+ }
+
+ @Override
+ public boolean isNormalizedKeyPrefixOnly(int keyBytes) {
+ return true;
+ }
+
+ @Override
+ public void putNormalizedKey(Tuple2<byte[], StreamRecord<IN>> record,
MemorySegment target, int offset, int numBytes) {
+ BytesKeyNormalizationUtil.putNormalizedKey(record,
record.f0.length, target, offset, numBytes);
+ }
+
+ @Override
+ public boolean invertNormalizedKey() {
+ return false;
+ }
+
+ @Override
+ public TypeComparator<Tuple2<byte[], StreamRecord<IN>>> duplicate() {
+ return new VariableLengthByteKeyComparator<>();
+ }
+
+ @Override
+ public int extractKeys(Object record, Object[] target, int index) {
+ target[index] = record;
+ return 1;
+ }
+
+ @Override
+ public TypeComparator<?>[] getFlatComparators() {
+ return new TypeComparator[] {this};
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // unsupported normalization
+ //
--------------------------------------------------------------------------------------------
+
+ @Override
+ public boolean supportsSerializationWithKeyNormalization() {
+ return false;
+ }
+
+ @Override
+ public void writeWithKeyNormalization(
+ Tuple2<byte[], StreamRecord<IN>> record,
+ DataOutputView target) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Tuple2<byte[], StreamRecord<IN>> readWithKeyDenormalization(
+ Tuple2<byte[], StreamRecord<IN>> reuse,
+ DataInputView source) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/FixedLengthByteKeyComparatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/FixedLengthByteKeyComparatorTest.java
new file mode 100644
index 0000000..6fe0fd3
--- /dev/null
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/FixedLengthByteKeyComparatorTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.operators.Order;
+import org.apache.flink.api.common.typeutils.ComparatorTestBase;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link FixedLengthByteKeyComparator}.
+ */
+public class FixedLengthByteKeyComparatorTest extends
ComparatorTestBase<Tuple2<byte[], StreamRecord<Integer>>> {
+ @Override
+ protected Order[] getTestedOrder() {
+ return new Order[]{Order.ASCENDING};
+ }
+
+ @Override
+ protected TypeComparator<Tuple2<byte[], StreamRecord<Integer>>>
createComparator(boolean ascending) {
+ return new FixedLengthByteKeyComparator<>(
+ new IntSerializer().getLength()
+ );
+ }
+
+ @Override
+ protected TypeSerializer<Tuple2<byte[], StreamRecord<Integer>>>
createSerializer() {
+ IntSerializer intSerializer = new IntSerializer();
+ return new KeyAndValueSerializer<>(
+ intSerializer,
+ intSerializer.getLength()
+ );
+ }
+
+ @Override
+ protected void deepEquals(
+ String message,
+ Tuple2<byte[], StreamRecord<Integer>> should,
+ Tuple2<byte[], StreamRecord<Integer>> is) {
+ assertThat(message, should.f0, equalTo(is.f0));
+ assertThat(message, should.f1, equalTo(is.f1));
+ }
+
+ @Override
+ protected Tuple2<byte[], StreamRecord<Integer>>[] getSortedTestData() {
+ return SerializerComparatorTestData.getOrderedIntTestData();
+ }
+}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/FixedLengthKeyAndValueSerializerTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/FixedLengthKeyAndValueSerializerTest.java
new file mode 100644
index 0000000..50322b4
--- /dev/null
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/FixedLengthKeyAndValueSerializerTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.junit.Test;
+
+/**
+ * Tests for {@link KeyAndValueSerializer}, which verify fixed length keys.
+ */
+public class FixedLengthKeyAndValueSerializerTest extends
SerializerTestBase<Tuple2<byte[], StreamRecord<Integer>>> {
+
+ private static final int INTEGER_SIZE = 4;
+ private static final int TIMESTAMP_SIZE = 8;
+
+ @Override
+ protected TypeSerializer<Tuple2<byte[], StreamRecord<Integer>>>
createSerializer() {
+ IntSerializer intSerializer = new IntSerializer();
+ return new KeyAndValueSerializer<>(
+ intSerializer,
+ intSerializer.getLength()
+ );
+ }
+
+ @Override
+ protected int getLength() {
+ return INTEGER_SIZE + TIMESTAMP_SIZE + INTEGER_SIZE;
+ }
+
+ @Override
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ protected Class<Tuple2<byte[], StreamRecord<Integer>>> getTypeClass() {
+ return (Class<Tuple2<byte[], StreamRecord<Integer>>>) (Class)
Tuple2.class;
+ }
+
+ @Override
+ protected Tuple2<byte[], StreamRecord<Integer>>[] getTestData() {
+ return SerializerComparatorTestData.getOrderedIntTestData();
+ }
+
+ @Override
+ @Test(expected = UnsupportedOperationException.class)
+ public void testConfigSnapshotInstantiation() {
+ super.testConfigSnapshotInstantiation();
+ }
+
+ @Override
+ @Test(expected = UnsupportedOperationException.class)
+ public void testSnapshotConfigurationAndReconfigure() throws Exception {
+ super.testSnapshotConfigurationAndReconfigure();
+ }
+}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SerializerComparatorTestData.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SerializerComparatorTestData.java
new file mode 100644
index 0000000..954d55f
--- /dev/null
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SerializerComparatorTestData.java
@@ -0,0 +1,109 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.io.IOException;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * Test data for {@link VariableLengthByteKeyComparatorTest}, {@link
FixedLengthByteKeyComparatorTest},
+ * {@link FixedLengthKeyAndValueSerializerTest}, and {@link
VariableLengthKeyAndValueSerializerTest}.
+ */
+final class SerializerComparatorTestData {
+ @SuppressWarnings("unchecked")
+ static Tuple2<byte[], StreamRecord<Integer>>[] getOrderedIntTestData() {
+ IntSerializer intSerializer = new IntSerializer();
+ DataOutputSerializer outputSerializer = new
DataOutputSerializer(intSerializer.getLength());
+
+ return IntStream.range(-10, 10)
+ .mapToObj(
+ idx -> {
+ try {
+ intSerializer.serialize(idx,
outputSerializer);
+ byte[] copyOfBuffer =
outputSerializer.getCopyOfBuffer();
+ outputSerializer.clear();
+ return Tuple2.of(copyOfBuffer,
new StreamRecord<>(idx, idx));
+ } catch (IOException e) {
+ throw new AssertionError(e);
+ }
+ }
+ ).toArray(Tuple2[]::new);
+ }
+
+ @SuppressWarnings("unchecked")
+ static Tuple2<byte[], StreamRecord<String>>[]
getOrderedStringTestData() {
+ StringSerializer stringSerializer = new StringSerializer();
+ DataOutputSerializer outputSerializer = new
DataOutputSerializer(64);
+ return Stream.of(
+ new String(new byte[] {-1, 0}),
+ new String(new byte[] {0, 1}),
+ "A",
+ "AB",
+ "ABC",
+ "ABCD",
+ "ABCDE",
+ "ABCDEF",
+ "ABCDEFG",
+ "ABCDEFGH")
+ .map(
+ str -> {
+ try {
+ stringSerializer.serialize(str,
outputSerializer);
+ byte[] copyOfBuffer =
outputSerializer.getCopyOfBuffer();
+ outputSerializer.clear();
+ return Tuple2.of(copyOfBuffer,
new StreamRecord<>(str, 0));
+ } catch (IOException e) {
+ throw new AssertionError(e);
+ }
+ }
+ ).sorted(
+ (o1, o2) -> {
+ byte[] key0 = o1.f0;
+ byte[] key1 = o2.f0;
+
+ int firstLength = key0.length;
+ int secondLength = key1.length;
+ int minLength = Math.min(firstLength,
secondLength);
+ for (int i = 0; i < minLength; i++) {
+ int cmp = Byte.compare(key0[i],
key1[i]);
+
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+
+ int lengthCmp =
Integer.compare(firstLength, secondLength);
+ if (lengthCmp != 0) {
+ return lengthCmp;
+ }
+ return
Long.compare(o1.f1.getTimestamp(), o2.f1.getTimestamp());
+ }
+ ).toArray(Tuple2[]::new);
+ }
+
+ private SerializerComparatorTestData() {
+ }
+}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortingDataInputITCase.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortingDataInputITCase.java
new file mode 100644
index 0000000..1e5b738
--- /dev/null
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortingDataInputITCase.java
@@ -0,0 +1,243 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import
org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.streamstatus.StreamStatus;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.LinkedHashSet;
+import java.util.Objects;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Longer running IT tests for {@link SortingDataInputTest}. For quicker smoke
tests see {@link SortingDataInputTest}.
+ */
+public class SortingDataInputITCase {
+ @Test
+ public void intKeySorting() throws Exception {
+ int numberOfRecords = 1_000_000;
+ GeneratedRecordsDataInput input = new
GeneratedRecordsDataInput(numberOfRecords);
+ KeySelector<Tuple3<Integer, String, byte[]>, Integer>
keySelector = value -> value.f0;
+ try (
+ MockEnvironment environment =
MockEnvironment.builder().build();
+ SortingDataInput<Tuple3<Integer, String, byte[]>,
Integer> sortingDataInput = new SortingDataInput<>(
+ input,
+ GeneratedRecordsDataInput.SERIALIZER,
+ new IntSerializer(),
+ keySelector,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable()
+ )) {
+ InputStatus inputStatus;
+ VerifyingOutput<Integer> output = new
VerifyingOutput<>(keySelector);
+ do {
+ inputStatus = sortingDataInput.emitNext(output);
+ } while (inputStatus != InputStatus.END_OF_INPUT);
+
+ assertThat(output.getSeenRecords(),
equalTo(numberOfRecords));
+ }
+ }
+
+ @Test
+ public void stringKeySorting() throws Exception {
+ int numberOfRecords = 1_000_000;
+ GeneratedRecordsDataInput input = new
GeneratedRecordsDataInput(numberOfRecords);
+ KeySelector<Tuple3<Integer, String, byte[]>, String>
keySelector = value -> value.f1;
+ try (
+ MockEnvironment environment =
MockEnvironment.builder().build();
+ SortingDataInput<Tuple3<Integer, String, byte[]>,
String> sortingDataInput = new SortingDataInput<>(
+ input,
+ GeneratedRecordsDataInput.SERIALIZER,
+ new StringSerializer(),
+ keySelector,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable()
+ )) {
+ InputStatus inputStatus;
+ VerifyingOutput<String> output = new
VerifyingOutput<>(keySelector);
+ do {
+ inputStatus = sortingDataInput.emitNext(output);
+ } while (inputStatus != InputStatus.END_OF_INPUT);
+
+ assertThat(output.getSeenRecords(),
equalTo(numberOfRecords));
+ }
+ }
+
+ /**
+ * The idea of the tests here is to check that the keys are grouped
together. Therefore there should not be a
+ * situation were we see a key different from the key of the previous
record, but one that we've seen before.
+ *
+ * <p>This output verifies that invariant.
+ */
+ private static final class VerifyingOutput<E> implements
PushingAsyncDataInput.DataOutput<Tuple3<Integer, String, byte[]>> {
+
+ private final KeySelector<Tuple3<Integer, String, byte[]>, E>
keySelector;
+ private final Set<E> seenKeys = new LinkedHashSet<>();
+ private E currentKey = null;
+ private int seenRecords = 0;
+
+ private VerifyingOutput(KeySelector<Tuple3<Integer, String,
byte[]>, E> keySelector) {
+ this.keySelector = keySelector;
+ }
+
+ @Override
+ public void emitRecord(StreamRecord<Tuple3<Integer, String,
byte[]>> streamRecord) throws Exception {
+ this.seenRecords++;
+ E incomingKey =
keySelector.getKey(streamRecord.getValue());
+ if (!Objects.equals(incomingKey, currentKey)) {
+ if (!seenKeys.add(incomingKey)) {
+ Assert.fail("Received an out of order
key: " + incomingKey);
+ }
+ this.currentKey = incomingKey;
+ }
+ }
+
+ @Override
+ public void emitWatermark(Watermark watermark) throws Exception
{
+
+ }
+
+ @Override
+ public void emitStreamStatus(StreamStatus streamStatus) throws
Exception {
+
+ }
+
+ @Override
+ public void emitLatencyMarker(LatencyMarker latencyMarker)
throws Exception {
+
+ }
+
+ public int getSeenRecords() {
+ return seenRecords;
+ }
+ }
+
+ private static final class GeneratedRecordsDataInput
+ implements StreamTaskInput<Tuple3<Integer, String, byte[]>> {
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static final TypeSerializer<Tuple3<Integer, String,
byte[]>> SERIALIZER = new TupleSerializer<>(
+ (Class<Tuple3<Integer, String, byte[]>>) (Class)
Tuple3.class,
+ new TypeSerializer[]{
+ new IntSerializer(),
+ new StringSerializer(),
+ new BytePrimitiveArraySerializer()
+ });
+
+ private static final String ALPHA_NUM =
"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
+ private final long numberOfRecords;
+ private int recordsGenerated;
+ private final Random rnd = new Random();
+ private final byte[] buffer;
+
+ private GeneratedRecordsDataInput(int numberOfRecords) {
+ this.numberOfRecords = numberOfRecords;
+ this.recordsGenerated = 0;
+ this.buffer = new byte[500];
+ rnd.nextBytes(buffer);
+ }
+
+ @Override
+ public InputStatus emitNext(DataOutput<Tuple3<Integer, String,
byte[]>> output) throws Exception {
+ if (recordsGenerated >= numberOfRecords) {
+ return InputStatus.END_OF_INPUT;
+ }
+
+ output.emitRecord(
+ new StreamRecord<>(
+ Tuple3.of(
+ rnd.nextInt(),
+ randomString(rnd.nextInt(256)),
+ buffer
+ ),
+ 1
+ )
+ );
+ if (recordsGenerated++ >= numberOfRecords) {
+ return InputStatus.END_OF_INPUT;
+ } else {
+ return InputStatus.MORE_AVAILABLE;
+ }
+ }
+
+ @Override
+ public CompletableFuture<?> getAvailableFuture() {
+ return AvailabilityProvider.AVAILABLE;
+ }
+
+ private String randomString(int len) {
+ StringBuilder sb = new StringBuilder(len);
+ for (int i = 0; i < len; i++) {
+
sb.append(ALPHA_NUM.charAt(rnd.nextInt(ALPHA_NUM.length())));
+ }
+ return sb.toString();
+ }
+
+ @Override
+ public int getInputIndex() {
+ return 0;
+ }
+
+ @Override
+ public CompletableFuture<Void> prepareSnapshot(
+ ChannelStateWriter channelStateWriter,
+ long checkpointId) throws IOException {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+ }
+}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortingDataInputTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortingDataInputTest.java
new file mode 100644
index 0000000..2712e8a
--- /dev/null
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortingDataInputTest.java
@@ -0,0 +1,268 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.streamstatus.StreamStatus;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link SortingDataInput}.
+ *
+ * <p>These are rather simple unit tests. See also {@link
SortingDataInputITCase} for more thorough tests.
+ */
+public class SortingDataInputTest {
+ @Test
+ public void simpleFixedLengthKeySorting() throws Exception {
+ CollectingDataOutput<Integer> collectingDataOutput = new
CollectingDataOutput<>();
+ CollectionDataInput<Integer> input = new CollectionDataInput<>(
+ Arrays.asList(
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 3),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(2, 2)
+ )
+ );
+ MockEnvironment environment = MockEnvironment.builder().build();
+ SortingDataInput<Integer, Integer> sortingDataInput = new
SortingDataInput<>(
+ input,
+ new IntSerializer(),
+ new IntSerializer(),
+ (KeySelector<Integer, Integer>) value -> value,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable()
+ );
+
+ InputStatus inputStatus;
+ do {
+ inputStatus =
sortingDataInput.emitNext(collectingDataOutput);
+ } while (inputStatus != InputStatus.END_OF_INPUT);
+
+ assertThat(collectingDataOutput.events, equalTo(
+ Arrays.asList(
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 2),
+ new StreamRecord<>(2, 3)
+ )
+ ));
+ }
+
+ @Test
+ public void watermarkPropagation() throws Exception {
+ CollectingDataOutput<Integer> collectingDataOutput = new
CollectingDataOutput<>();
+ CollectionDataInput<Integer> input = new CollectionDataInput<>(
+ Arrays.asList(
+ new StreamRecord<>(1, 3),
+ new Watermark(1),
+ new StreamRecord<>(1, 1),
+ new Watermark(2),
+ new StreamRecord<>(2, 1),
+ new Watermark(3),
+ new StreamRecord<>(2, 3),
+ new Watermark(4),
+ new StreamRecord<>(1, 2),
+ new Watermark(5),
+ new StreamRecord<>(2, 2),
+ new Watermark(6)
+ )
+ );
+ MockEnvironment environment = MockEnvironment.builder().build();
+ SortingDataInput<Integer, Integer> sortingDataInput = new
SortingDataInput<>(
+ input,
+ new IntSerializer(),
+ new IntSerializer(),
+ (KeySelector<Integer, Integer>) value -> value,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable()
+ );
+
+ InputStatus inputStatus;
+ do {
+ inputStatus =
sortingDataInput.emitNext(collectingDataOutput);
+ } while (inputStatus != InputStatus.END_OF_INPUT);
+
+ assertThat(collectingDataOutput.events, equalTo(
+ Arrays.asList(
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 2),
+ new StreamRecord<>(2, 3),
+ new Watermark(6)
+ )
+ ));
+ }
+
+ @Test
+ public void simpleVariableLengthKeySorting() throws Exception {
+ CollectingDataOutput<Integer> collectingDataOutput = new
CollectingDataOutput<>();
+ CollectionDataInput<Integer> input = new CollectionDataInput<>(
+ Arrays.asList(
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 3),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(2, 2)
+ )
+ );
+ MockEnvironment environment = MockEnvironment.builder().build();
+ SortingDataInput<Integer, String> sortingDataInput = new
SortingDataInput<>(
+ input,
+ new IntSerializer(),
+ new StringSerializer(),
+ (KeySelector<Integer, String>) value -> "" + value,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable()
+ );
+
+ InputStatus inputStatus;
+ do {
+ inputStatus =
sortingDataInput.emitNext(collectingDataOutput);
+ } while (inputStatus != InputStatus.END_OF_INPUT);
+
+ assertThat(collectingDataOutput.events, equalTo(
+ Arrays.asList(
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 2),
+ new StreamRecord<>(2, 3)
+ )
+ ));
+ }
+
+ private static final class CollectionDataInput<E> implements
StreamTaskInput<E> {
+
+ private final Iterator<StreamElement> elementsIterator;
+
+ private CollectionDataInput(Collection<StreamElement> elements)
{
+ this.elementsIterator = elements.iterator();
+ }
+
+ @Override
+ public InputStatus emitNext(DataOutput<E> output) throws
Exception {
+ if (elementsIterator.hasNext()) {
+ StreamElement streamElement =
elementsIterator.next();
+ if (streamElement instanceof StreamRecord) {
+
output.emitRecord(streamElement.asRecord());
+ } else if (streamElement instanceof Watermark) {
+
output.emitWatermark(streamElement.asWatermark());
+ } else {
+ throw new
IllegalStateException("Unsupported element type: " + streamElement);
+ }
+ }
+ return elementsIterator.hasNext() ?
InputStatus.MORE_AVAILABLE : InputStatus.END_OF_INPUT;
+ }
+
+ @Override
+ public CompletableFuture<?> getAvailableFuture() {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public int getInputIndex() {
+ return 0;
+ }
+
+ @Override
+ public CompletableFuture<Void> prepareSnapshot(
+ ChannelStateWriter channelStateWriter,
+ long checkpointId) throws IOException {
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+ }
+
+ /**
+ * A test utility implementation of {@link
PushingAsyncDataInput.DataOutput} that collects all events.
+ */
+ private static final class CollectingDataOutput<E> implements
PushingAsyncDataInput.DataOutput<E> {
+
+ final List<Object> events = new ArrayList<>();
+
+ @Override
+ public void emitWatermark(Watermark watermark) throws Exception
{
+ events.add(watermark);
+ }
+
+ @Override
+ public void emitStreamStatus(StreamStatus streamStatus) throws
Exception {
+ events.add(streamStatus);
+ }
+
+ @Override
+ public void emitRecord(StreamRecord<E> streamRecord) throws
Exception {
+ events.add(streamRecord);
+ }
+
+ @Override
+ public void emitLatencyMarker(LatencyMarker latencyMarker)
throws Exception {
+ events.add(latencyMarker);
+ }
+ }
+}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/VariableLengthByteKeyComparatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/VariableLengthByteKeyComparatorTest.java
new file mode 100644
index 0000000..05848a2
--- /dev/null
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/VariableLengthByteKeyComparatorTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.operators.Order;
+import org.apache.flink.api.common.typeutils.ComparatorTestBase;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link VariableLengthByteKeyComparator}.
+ */
+public class VariableLengthByteKeyComparatorTest extends
ComparatorTestBase<Tuple2<byte[], StreamRecord<String>>> {
+ @Override
+ protected Order[] getTestedOrder() {
+ return new Order[]{Order.ASCENDING};
+ }
+
+ @Override
+ protected TypeComparator<Tuple2<byte[], StreamRecord<String>>>
createComparator(boolean ascending) {
+ return new VariableLengthByteKeyComparator<>();
+ }
+
+ @Override
+ protected TypeSerializer<Tuple2<byte[], StreamRecord<String>>>
createSerializer() {
+ StringSerializer stringSerializer = new StringSerializer();
+ return new KeyAndValueSerializer<>(
+ stringSerializer,
+ stringSerializer.getLength()
+ );
+ }
+
+ @Override
+ protected void deepEquals(
+ String message,
+ Tuple2<byte[], StreamRecord<String>> should,
+ Tuple2<byte[], StreamRecord<String>> is) {
+ assertThat(message, should.f0, equalTo(is.f0));
+ assertThat(message, should.f1, equalTo(is.f1));
+ }
+
+ @Override
+ protected Tuple2<byte[], StreamRecord<String>>[] getSortedTestData() {
+ return SerializerComparatorTestData.getOrderedStringTestData();
+ }
+}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/VariableLengthKeyAndValueSerializerTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/VariableLengthKeyAndValueSerializerTest.java
new file mode 100644
index 0000000..1758657
--- /dev/null
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/VariableLengthKeyAndValueSerializerTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.api.operators.sort;
+
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.junit.Test;
+
+/**
+ * Tests for {@link KeyAndValueSerializer}, which verify variable length keys.
+ */
+public class VariableLengthKeyAndValueSerializerTest extends
SerializerTestBase<Tuple2<byte[], StreamRecord<String>>> {
+
+ @Override
+ protected TypeSerializer<Tuple2<byte[], StreamRecord<String>>>
createSerializer() {
+ StringSerializer stringSerializer = new StringSerializer();
+ return new KeyAndValueSerializer<>(
+ stringSerializer,
+ stringSerializer.getLength()
+ );
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ protected Class<Tuple2<byte[], StreamRecord<String>>> getTypeClass() {
+ return (Class<Tuple2<byte[], StreamRecord<String>>>) (Class)
Tuple2.class;
+ }
+
+ @Override
+ protected Tuple2<byte[], StreamRecord<String>>[] getTestData() {
+ return SerializerComparatorTestData.getOrderedStringTestData();
+ }
+
+ @Override
+ @Test(expected = UnsupportedOperationException.class)
+ public void testConfigSnapshotInstantiation() {
+ super.testConfigSnapshotInstantiation();
+ }
+
+ @Override
+ @Test(expected = UnsupportedOperationException.class)
+ public void testSnapshotConfigurationAndReconfigure() throws Exception {
+ super.testSnapshotConfigurationAndReconfigure();
+ }
+}
