You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2020/04/24 15:30:29 UTC
[flink] branch master updated: [FLINK-13639] Refactor the
IntermediateResultPartitionID to consist of IntermediateDataSetID and
partitionIndex
This is an automated email from the ASF dual-hosted git repository.
zhuzh 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 c4b44e9 [FLINK-13639] Refactor the IntermediateResultPartitionID to consist of IntermediateDataSetID and partitionIndex
c4b44e9 is described below
commit c4b44e9d4a7af9e841f785fcf0e4831f4cd5afe9
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Wed Apr 22 14:52:21 2020 +0800
[FLINK-13639] Refactor the IntermediateResultPartitionID to consist of IntermediateDataSetID and partitionIndex
---
.../IntermediateResultPartition.java | 2 +-
.../runtime/io/network/netty/NettyMessage.java | 4 +-
.../runtime/jobgraph/IntermediateDataSetID.java | 17 +++++++
.../jobgraph/IntermediateResultPartitionID.java | 54 +++++++++++++++++-----
.../apache/flink/runtime/topology/ResultID.java | 2 +-
5 files changed, 64 insertions(+), 15 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java
index 5cdd376b..78bd9c4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/IntermediateResultPartition.java
@@ -46,7 +46,7 @@ public class IntermediateResultPartition {
this.producer = producer;
this.partitionNumber = partitionNumber;
this.consumers = new ArrayList<List<ExecutionEdge>>(0);
- this.partitionId = new IntermediateResultPartitionID();
+ this.partitionId = new IntermediateResultPartitionID(totalResult.getId(), partitionNumber);
}
public ExecutionVertex getProducer() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
index 2ed875d..5523cf2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyMessage.java
@@ -508,7 +508,7 @@ public abstract class NettyMessage {
ByteBuf result = null;
try {
- result = allocateBuffer(allocator, ID, 16 + 16 + 4 + 16 + 4);
+ result = allocateBuffer(allocator, ID, 20 + 16 + 4 + 16 + 4);
partitionId.getPartitionId().writeTo(result);
partitionId.getProducerId().writeTo(result);
@@ -569,7 +569,7 @@ public abstract class NettyMessage {
// TODO Directly serialize to Netty's buffer
ByteBuffer serializedEvent = EventSerializer.toSerializedEvent(event);
- result = allocateBuffer(allocator, ID, 4 + serializedEvent.remaining() + 16 + 16 + 16);
+ result = allocateBuffer(allocator, ID, 4 + serializedEvent.remaining() + 20 + 16 + 16);
result.writeInt(serializedEvent.remaining());
result.writeBytes(serializedEvent);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSetID.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSetID.java
index c22ca1e..b34e3f3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSetID.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSetID.java
@@ -21,6 +21,8 @@ package org.apache.flink.runtime.jobgraph;
import org.apache.flink.runtime.topology.ResultID;
import org.apache.flink.util.AbstractID;
+import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
+
import java.util.UUID;
/**
@@ -54,4 +56,19 @@ public class IntermediateDataSetID extends AbstractID implements ResultID {
public IntermediateDataSetID(UUID from) {
super(from.getLeastSignificantBits(), from.getMostSignificantBits());
}
+
+ private IntermediateDataSetID(long lower, long upper) {
+ super(lower, upper);
+ }
+
+ public void writeTo(ByteBuf buf) {
+ buf.writeLong(lowerPart);
+ buf.writeLong(upperPart);
+ }
+
+ public static IntermediateDataSetID fromByteBuf(ByteBuf buf) {
+ final long lower = buf.readLong();
+ final long upper = buf.readLong();
+ return new IntermediateDataSetID(lower, upper);
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateResultPartitionID.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateResultPartitionID.java
index 803c000..f43a5b5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateResultPartitionID.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateResultPartitionID.java
@@ -18,38 +18,70 @@
package org.apache.flink.runtime.jobgraph;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.executiongraph.IntermediateResultPartition;
import org.apache.flink.runtime.topology.ResultID;
-import org.apache.flink.util.AbstractID;
import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
/**
* Id identifying {@link IntermediateResultPartition}.
*/
-public class IntermediateResultPartitionID extends AbstractID implements ResultID {
+public class IntermediateResultPartitionID implements ResultID {
private static final long serialVersionUID = 1L;
+ private final IntermediateDataSetID intermediateDataSetID;
+ private final int partitionNum;
+
/**
- * Creates an new random intermediate result partition ID.
+ * Creates an new random intermediate result partition ID for testing.
*/
+ @VisibleForTesting
public IntermediateResultPartitionID() {
- super();
+ this.partitionNum = -1;
+ this.intermediateDataSetID = new IntermediateDataSetID();
}
- public IntermediateResultPartitionID(long lowerPart, long upperPart) {
- super(lowerPart, upperPart);
+ /**
+ * Creates an new intermediate result partition ID with {@link IntermediateDataSetID} and the partitionNum.
+ */
+ public IntermediateResultPartitionID(IntermediateDataSetID intermediateDataSetID, int partitionNum) {
+ this.intermediateDataSetID = intermediateDataSetID;
+ this.partitionNum = partitionNum;
}
public void writeTo(ByteBuf buf) {
- buf.writeLong(this.lowerPart);
- buf.writeLong(this.upperPart);
+ intermediateDataSetID.writeTo(buf);
+ buf.writeInt(partitionNum);
}
public static IntermediateResultPartitionID fromByteBuf(ByteBuf buf) {
- long lower = buf.readLong();
- long upper = buf.readLong();
- return new IntermediateResultPartitionID(lower, upper);
+ final IntermediateDataSetID intermediateDataSetID = IntermediateDataSetID.fromByteBuf(buf);
+ final int partitionNum = buf.readInt();
+ return new IntermediateResultPartitionID(intermediateDataSetID, partitionNum);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ } else if (obj != null && obj.getClass() == getClass()) {
+ IntermediateResultPartitionID that = (IntermediateResultPartitionID) obj;
+ return that.intermediateDataSetID.equals(this.intermediateDataSetID)
+ && that.partitionNum == this.partitionNum;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public int hashCode() {
+ return this.intermediateDataSetID.hashCode() ^ this.partitionNum;
+ }
+
+ @Override
+ public String toString() {
+ return intermediateDataSetID.toString() + "#" + partitionNum;
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/topology/ResultID.java b/flink-runtime/src/main/java/org/apache/flink/runtime/topology/ResultID.java
index 606544f..e616316 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/topology/ResultID.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/topology/ResultID.java
@@ -21,5 +21,5 @@ package org.apache.flink.runtime.topology;
/**
* ID of a {@link Result}.
*/
-public interface ResultID {
+public interface ResultID extends java.io.Serializable {
}