You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2023/05/11 01:36:50 UTC
[flink] branch master updated (88b3432a284 -> 80a924309ce)
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 88b3432a284 [FLINK-32029][core] Adds fallback error handling to FutureUtils.handleUncaughtException.
new e70c03495d5 [FLINK-31635][network] Introduce the identifiers in the tiered storage
new 14707742d24 [FLINK-31635][network] Introduce the tiered storage configuration
new 8a903fa0a30 [FLINK-31635][network] Introduce the interfaces in the tier for the tiered storage
new bd11c4036ef [FLINK-31635][network] Introduce tiered storage resource and the resource registry
new 7aba50fe56a [FLINK-31635][network] Introduce the multiple storage clients for the tiered storage
new 80a924309ce [FLINK-31635][network] Introduce tiered result partition
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../io/network/partition/ResultPartitionID.java | 23 +++
.../TieredStorageBytesBasedDataIdentifier.java | 76 +++++++
.../tiered/common/TieredStorageConfiguration.java | 53 +++++
.../tiered/common/TieredStorageDataIdentifier.java | 22 ++
.../tiered/common/TieredStorageIdMappingUtils.java | 51 +++++
.../tiered/common/TieredStoragePartitionId.java | 40 ++++
.../tiered/common/TieredStorageSubpartitionId.java | 65 ++++++
.../hybrid/tiered/common/TieredStorageTopicId.java | 40 ++++
.../shuffle/TieredInternalShuffleMaster.java | 97 +++++++++
.../tiered/shuffle/TieredResultPartition.java | 225 +++++++++++++++++++++
.../hybrid/tiered/storage/BufferAccumulator.java | 59 ++++++
.../storage/TieredStorageConsumerClient.java | 22 ++
.../tiered/storage/TieredStorageMasterClient.java | 42 ++++
.../storage/TieredStorageProducerClient.java | 125 ++++++++++++
.../tiered/storage/TieredStorageResource.java | 26 +++
.../storage/TieredStorageResourceRegistry.java | 62 ++++++
.../hybrid/tiered/tier/TierConsumerAgent.java | 22 ++
.../partition/hybrid/tiered/tier/TierFactory.java | 34 ++++
.../hybrid/tiered/tier/TierMasterAgent.java | 39 ++++
.../hybrid/tiered/tier/TierProducerAgent.java | 49 +++++
.../flink/runtime/shuffle/NettyShuffleMaster.java | 15 +-
.../hybrid/tiered/TestingBufferAccumulator.java | 45 +++++
.../common/TieredStorageIdMappingUtilsTest.java | 58 ++++++
.../tiered/shuffle/TieredResultPartitionTest.java | 216 ++++++++++++++++++++
24 files changed, 1505 insertions(+), 1 deletion(-)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageBytesBasedDataIdentifier.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageDataIdentifier.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageIdMappingUtils.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStoragePartitionId.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageSubpartitionId.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageTopicId.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredInternalShuffleMaster.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartition.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/BufferAccumulator.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageConsumerClient.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMasterClient.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageResource.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageResourceRegistry.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierConsumerAgent.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierMasterAgent.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingBufferAccumulator.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageIdMappingUtilsTest.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartitionTest.java
[flink] 03/06: [FLINK-31635][network] Introduce the interfaces in the tier for the tiered storage
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8a903fa0a30a8717653426a4fa5ac0229c5c08b4
Author: Yuxin Tan <ta...@gmail.com>
AuthorDate: Wed May 10 19:47:37 2023 +0800
[FLINK-31635][network] Introduce the interfaces in the tier for the tiered storage
---
.../tiered/common/TieredStorageConfiguration.java | 11 +++++
.../TierConsumerAgent.java} | 26 ++----------
.../TierFactory.java} | 28 ++++---------
.../TierMasterAgent.java} | 35 +++++++---------
.../hybrid/tiered/tier/TierProducerAgent.java | 49 ++++++++++++++++++++++
5 files changed, 88 insertions(+), 61 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
index 185c267bd6f..e0d40c40c2d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
@@ -19,10 +19,21 @@
package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierFactory;
+
+import java.util.Arrays;
+import java.util.List;
/** Configurations for the Tiered Storage. */
public class TieredStorageConfiguration {
+ // TODO, after implementing the tier factory, add appreciate implementations to the array.
+ private static final TierFactory[] DEFAULT_MEMORY_DISK_TIER_FACTORIES = new TierFactory[0];
+
+ public List<TierFactory> getTierFactories() {
+ return Arrays.asList(DEFAULT_MEMORY_DISK_TIER_FACTORIES);
+ }
+
public static TieredStorageConfiguration.Builder builder() {
return new TieredStorageConfiguration.Builder();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierConsumerAgent.java
similarity index 51%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierConsumerAgent.java
index 185c267bd6f..b9dc497089c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierConsumerAgent.java
@@ -16,27 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
+package org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier;
-import org.apache.flink.configuration.Configuration;
-
-/** Configurations for the Tiered Storage. */
-public class TieredStorageConfiguration {
-
- public static TieredStorageConfiguration.Builder builder() {
- return new TieredStorageConfiguration.Builder();
- }
-
- /** Builder for {@link TieredStorageConfiguration}. */
- public static class Builder {
- public TieredStorageConfiguration build() {
- return new TieredStorageConfiguration();
- }
- }
-
- public static TieredStorageConfiguration fromConfiguration(Configuration conf) {
- // TODO, from the configuration, get the configured options(i.e., remote storage path, the
- // reserved storage size, etc.), then set them to the builder.
- return new TieredStorageConfiguration.Builder().build();
- }
-}
+/** The consumer-side agent of a Tier. */
+public interface TierConsumerAgent {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
similarity index 51%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
index 185c267bd6f..3225c3a5a09 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
@@ -16,27 +16,17 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
+package org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier;
-import org.apache.flink.configuration.Configuration;
+/** A factory that creates all the components of a tier. */
+public interface TierFactory {
-/** Configurations for the Tiered Storage. */
-public class TieredStorageConfiguration {
+ /** Creates the master-side agent of a Tier. */
+ TierMasterAgent createMasterAgent();
- public static TieredStorageConfiguration.Builder builder() {
- return new TieredStorageConfiguration.Builder();
- }
+ /** Creates the producer-side agent of a Tier. */
+ TierProducerAgent createProducerAgent();
- /** Builder for {@link TieredStorageConfiguration}. */
- public static class Builder {
- public TieredStorageConfiguration build() {
- return new TieredStorageConfiguration();
- }
- }
-
- public static TieredStorageConfiguration fromConfiguration(Configuration conf) {
- // TODO, from the configuration, get the configured options(i.e., remote storage path, the
- // reserved storage size, etc.), then set them to the builder.
- return new TieredStorageConfiguration.Builder().build();
- }
+ /** Creates the consumer-side agent of a Tier. */
+ TierConsumerAgent createConsumerAgent();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierMasterAgent.java
similarity index 51%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierMasterAgent.java
index 185c267bd6f..2c34cdd38c6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierMasterAgent.java
@@ -16,27 +16,24 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.io.network.partition.hybrid.tiered.common;
+package org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
-/** Configurations for the Tiered Storage. */
-public class TieredStorageConfiguration {
+/** The master-side agent of a Tier. */
+public interface TierMasterAgent {
- public static TieredStorageConfiguration.Builder builder() {
- return new TieredStorageConfiguration.Builder();
- }
+ /**
+ * Add a new tiered storage partition.
+ *
+ * @param partitionId the identifier of the new partition
+ */
+ void addPartition(TieredStoragePartitionId partitionId);
- /** Builder for {@link TieredStorageConfiguration}. */
- public static class Builder {
- public TieredStorageConfiguration build() {
- return new TieredStorageConfiguration();
- }
- }
-
- public static TieredStorageConfiguration fromConfiguration(Configuration conf) {
- // TODO, from the configuration, get the configured options(i.e., remote storage path, the
- // reserved storage size, etc.), then set them to the builder.
- return new TieredStorageConfiguration.Builder().build();
- }
+ /**
+ * Release a tiered storage partition.
+ *
+ * @param partitionId the identifier of partition to be released
+ */
+ void releasePartition(TieredStoragePartitionId partitionId);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java
new file mode 100644
index 00000000000..9c39e8e05ef
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java
@@ -0,0 +1,49 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.tier;
+
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+
+import java.io.IOException;
+
+/** The producer-side agent of a Tier. */
+public interface TierProducerAgent {
+
+ /**
+ * Try to start a new segment in the Tier.
+ *
+ * @param subpartitionId subpartition id that the new segment belongs to
+ * @param segmentId id of the new segment
+ * @return true if the segment can be started, false otherwise.
+ */
+ boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId);
+
+ /** Writes the finished {@link Buffer} to the consumer. */
+ boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer)
+ throws IOException;
+
+ /**
+ * Close the agent.
+ *
+ * <p>Note this only releases resources directly hold by the agent, which excludes resources
+ * managed by the resource registry.
+ */
+ void close();
+}
[flink] 01/06: [FLINK-31635][network] Introduce the identifiers in the tiered storage
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e70c03495d528a067b166c3de38cd9258f68a6fa
Author: Yuxin Tan <ta...@gmail.com>
AuthorDate: Mon May 8 13:46:40 2023 +0800
[FLINK-31635][network] Introduce the identifiers in the tiered storage
---
.../io/network/partition/ResultPartitionID.java | 23 +++++++
.../TieredStorageBytesBasedDataIdentifier.java | 76 ++++++++++++++++++++++
.../tiered/common/TieredStorageDataIdentifier.java | 22 +++++++
.../tiered/common/TieredStorageIdMappingUtils.java | 51 +++++++++++++++
.../tiered/common/TieredStoragePartitionId.java | 40 ++++++++++++
.../tiered/common/TieredStorageSubpartitionId.java | 65 ++++++++++++++++++
.../hybrid/tiered/common/TieredStorageTopicId.java | 40 ++++++++++++
.../common/TieredStorageIdMappingUtilsTest.java | 58 +++++++++++++++++
8 files changed, 375 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionID.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionID.java
index 5355ab3992e..116d34653dd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionID.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionID.java
@@ -23,6 +23,9 @@ import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.IntermediateResultPartition;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBuf;
+import org.apache.flink.shaded.netty4.io.netty.buffer.Unpooled;
+
import java.io.Serializable;
import static org.apache.flink.runtime.executiongraph.ExecutionAttemptID.randomId;
@@ -54,6 +57,15 @@ public final class ResultPartitionID implements Serializable {
this.producerId = checkNotNull(producerId);
}
+ public ResultPartitionID(byte[] bytes) {
+ ByteBuf byteBuf = Unpooled.buffer();
+ byteBuf.writeBytes(bytes);
+
+ this.partitionId = IntermediateResultPartitionID.fromByteBuf(byteBuf);
+ this.producerId = ExecutionAttemptID.fromByteBuf(byteBuf);
+ byteBuf.release();
+ }
+
public IntermediateResultPartitionID getPartitionId() {
return partitionId;
}
@@ -82,4 +94,15 @@ public final class ResultPartitionID implements Serializable {
public String toString() {
return partitionId.toString() + "@" + producerId.toString();
}
+
+ public byte[] getBytes() {
+ ByteBuf byteBuf = Unpooled.buffer();
+ partitionId.writeTo(byteBuf);
+ producerId.writeTo(byteBuf);
+
+ byte[] bytes = new byte[byteBuf.readableBytes()];
+ byteBuf.readBytes(bytes);
+ byteBuf.release();
+ return bytes;
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageBytesBasedDataIdentifier.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageBytesBasedDataIdentifier.java
new file mode 100644
index 00000000000..779530236ea
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageBytesBasedDataIdentifier.java
@@ -0,0 +1,76 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.common;
+
+import org.apache.flink.util.StringUtils;
+
+import java.io.Serializable;
+import java.util.Arrays;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** The bytes based unique identification for the Tiered Storage. */
+public abstract class TieredStorageBytesBasedDataIdentifier
+ implements TieredStorageDataIdentifier, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The bytes data of this identifier. */
+ protected final byte[] bytes;
+
+ protected final int hashCode;
+
+ public TieredStorageBytesBasedDataIdentifier(byte[] bytes) {
+ checkArgument(bytes != null, "Must be not null.");
+
+ this.bytes = bytes;
+ this.hashCode = Arrays.hashCode(bytes);
+ }
+
+ public byte[] getBytes() {
+ return bytes;
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (this == that) {
+ return true;
+ }
+
+ if (that == null || getClass() != that.getClass()) {
+ return false;
+ }
+
+ TieredStorageBytesBasedDataIdentifier thatID = (TieredStorageBytesBasedDataIdentifier) that;
+ return hashCode == thatID.hashCode && Arrays.equals(bytes, thatID.bytes);
+ }
+
+ @Override
+ public int hashCode() {
+ return hashCode;
+ }
+
+ @Override
+ public String toString() {
+ return "TieredStorageBytesBasedDataIdentifier{"
+ + "ID="
+ + StringUtils.byteToHexString(bytes)
+ + '}';
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageDataIdentifier.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageDataIdentifier.java
new file mode 100644
index 00000000000..b07e55f214d
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageDataIdentifier.java
@@ -0,0 +1,22 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.common;
+
+/** Identifier interface in the Tiered Storage. */
+public interface TieredStorageDataIdentifier {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageIdMappingUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageIdMappingUtils.java
new file mode 100644
index 00000000000..21e44719875
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageIdMappingUtils.java
@@ -0,0 +1,51 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.common;
+
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.util.AbstractID;
+
+/** Utils to convert the Ids to Tiered Storage Ids, or vice versa. */
+public class TieredStorageIdMappingUtils {
+
+ public static TieredStorageTopicId convertId(IntermediateDataSetID intermediateDataSetID) {
+ return new TieredStorageTopicId(intermediateDataSetID.getBytes());
+ }
+
+ public static IntermediateDataSetID convertId(TieredStorageTopicId topicId) {
+ return new IntermediateDataSetID(new AbstractID(topicId.getBytes()));
+ }
+
+ public static TieredStoragePartitionId convertId(ResultPartitionID resultPartitionId) {
+ return new TieredStoragePartitionId(resultPartitionId.getBytes());
+ }
+
+ public static ResultPartitionID convertId(TieredStoragePartitionId partitionId) {
+ return new ResultPartitionID(partitionId.getBytes());
+ }
+
+ public static TieredStorageSubpartitionId convertId(int subpartitionId) {
+ return new TieredStorageSubpartitionId(subpartitionId);
+ }
+
+ public static int convertId(TieredStorageSubpartitionId subpartitionId) {
+ return subpartitionId.getSubpartitionId();
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStoragePartitionId.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStoragePartitionId.java
new file mode 100644
index 00000000000..0d130f59bee
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStoragePartitionId.java
@@ -0,0 +1,40 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.common;
+
+import org.apache.flink.util.StringUtils;
+
+/**
+ * Identifier of a partition.
+ *
+ * <p>A partition is equivalent to a result partition in Flink.
+ */
+public class TieredStoragePartitionId extends TieredStorageBytesBasedDataIdentifier {
+
+ private static final long serialVersionUID = 1L;
+
+ public TieredStoragePartitionId(byte[] bytes) {
+ super(bytes);
+ }
+
+ @Override
+ public String toString() {
+ return "TieredStoragePartitionId{" + "ID=" + StringUtils.byteToHexString(bytes) + '}';
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageSubpartitionId.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageSubpartitionId.java
new file mode 100644
index 00000000000..2cb0f4f155f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageSubpartitionId.java
@@ -0,0 +1,65 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.common;
+
+import java.io.Serializable;
+
+/**
+ * Identifier of a subpartition.
+ *
+ * <p>A subpartition is equivalent to a subpartition in Flink.
+ */
+public class TieredStorageSubpartitionId implements TieredStorageDataIdentifier, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int subpartitionId;
+
+ public TieredStorageSubpartitionId(int subpartitionId) {
+ this.subpartitionId = subpartitionId;
+ }
+
+ public int getSubpartitionId() {
+ return subpartitionId;
+ }
+
+ @Override
+ public int hashCode() {
+ return subpartitionId;
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (this == that) {
+ return true;
+ }
+
+ if (that == null || getClass() != that.getClass()) {
+ return false;
+ }
+
+ TieredStorageSubpartitionId thatID = (TieredStorageSubpartitionId) that;
+ return subpartitionId == thatID.subpartitionId;
+ }
+
+ @Override
+ public String toString() {
+ return "TieredStorageSubpartitionId{" + "ID=" + subpartitionId + '}';
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageTopicId.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageTopicId.java
new file mode 100644
index 00000000000..f686a877ee8
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageTopicId.java
@@ -0,0 +1,40 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.common;
+
+import org.apache.flink.util.StringUtils;
+
+/**
+ * Identifier of a topic.
+ *
+ * <p>A topic is equivalent to an intermediate data set in Flink.
+ */
+public class TieredStorageTopicId extends TieredStorageBytesBasedDataIdentifier {
+
+ private static final long serialVersionUID = 1L;
+
+ public TieredStorageTopicId(byte[] bytes) {
+ super(bytes);
+ }
+
+ @Override
+ public String toString() {
+ return "TieredStorageTopicId{" + "ID=" + StringUtils.byteToHexString(bytes) + '}';
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageIdMappingUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageIdMappingUtilsTest.java
new file mode 100644
index 00000000000..f8363969af5
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageIdMappingUtilsTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.common;
+
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TieredStorageIdMappingUtils}. */
+public class TieredStorageIdMappingUtilsTest {
+
+ @Test
+ void testConvertDataSetId() {
+ IntermediateDataSetID dataSetID = new IntermediateDataSetID();
+ TieredStorageTopicId topicId = TieredStorageIdMappingUtils.convertId(dataSetID);
+ IntermediateDataSetID convertedDataSetID = TieredStorageIdMappingUtils.convertId(topicId);
+ assertThat(dataSetID).isEqualTo(convertedDataSetID);
+ }
+
+ @Test
+ void testConvertResultPartitionId() {
+ ResultPartitionID resultPartitionID = new ResultPartitionID();
+ TieredStoragePartitionId tieredStoragePartitionId =
+ TieredStorageIdMappingUtils.convertId(resultPartitionID);
+ ResultPartitionID convertedResultPartitionID =
+ TieredStorageIdMappingUtils.convertId(tieredStoragePartitionId);
+ assertThat(resultPartitionID).isEqualTo(convertedResultPartitionID);
+ }
+
+ @Test
+ void testConvertSubpartitionId() {
+ int subpartitionId = 2;
+ TieredStorageSubpartitionId tieredStorageSubpartitionId =
+ TieredStorageIdMappingUtils.convertId(subpartitionId);
+ int convertedSubpartitionId =
+ TieredStorageIdMappingUtils.convertId(tieredStorageSubpartitionId);
+ assertThat(subpartitionId).isEqualTo(convertedSubpartitionId);
+ }
+}
[flink] 05/06: [FLINK-31635][network] Introduce the multiple storage clients for the tiered storage
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7aba50fe56aa8b6b6ef2474ed1856aa87d92242c
Author: Yuxin Tan <ta...@gmail.com>
AuthorDate: Mon May 8 13:51:56 2023 +0800
[FLINK-31635][network] Introduce the multiple storage clients for the tiered storage
---
.../hybrid/tiered/storage/BufferAccumulator.java | 59 ++++++++++
.../storage/TieredStorageConsumerClient.java | 22 ++++
.../tiered/storage/TieredStorageMasterClient.java | 42 +++++++
.../storage/TieredStorageProducerClient.java | 125 +++++++++++++++++++++
4 files changed, 248 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/BufferAccumulator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/BufferAccumulator.java
new file mode 100644
index 00000000000..e57f8383051
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/BufferAccumulator.java
@@ -0,0 +1,59 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.storage;
+
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.BiConsumer;
+
+/**
+ * Accumulates received records into buffers. The {@link BufferAccumulator} receives the records
+ * from tiered store producer and the records will accumulate and transform into buffers.
+ */
+public interface BufferAccumulator {
+
+ /**
+ * Setup the accumulator.
+ *
+ * @param numSubpartitions number of subpartitions
+ * @param bufferFlusher accepts the accumulated buffers. The first field is the subpartition id,
+ * while the list in the second field contains accumulated buffers in order for that
+ * subpartition.
+ */
+ void setup(
+ int numSubpartitions,
+ BiConsumer<TieredStorageSubpartitionId, List<Buffer>> bufferFlusher);
+
+ /**
+ * Receives the records from tiered store producer, these records will be accumulated and
+ * transformed into finished buffers.
+ */
+ void receive(
+ ByteBuffer record, TieredStorageSubpartitionId subpartitionId, Buffer.DataType dataType)
+ throws IOException;
+
+ /**
+ * Close the accumulator. This will flush all the remaining data and release all the resources.
+ */
+ void close();
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageConsumerClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageConsumerClient.java
new file mode 100644
index 00000000000..f1b69c347dc
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageConsumerClient.java
@@ -0,0 +1,22 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.storage;
+
+/** Client of the Tiered Storage used by the consumer. */
+public class TieredStorageConsumerClient {}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMasterClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMasterClient.java
new file mode 100644
index 00000000000..60f2c779f30
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageMasterClient.java
@@ -0,0 +1,42 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.storage;
+
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierMasterAgent;
+
+import java.util.List;
+
+/** Client of the Tiered Storage used by the master. */
+public class TieredStorageMasterClient {
+
+ private final List<TierMasterAgent> tiers;
+
+ public TieredStorageMasterClient(List<TierMasterAgent> tiers) {
+ this.tiers = tiers;
+ }
+
+ public void addPartition(TieredStoragePartitionId partitionId) {
+ tiers.forEach(tierMasterAgent -> tierMasterAgent.addPartition(partitionId));
+ }
+
+ public void releasePartition(TieredStoragePartitionId partitionId) {
+ tiers.forEach(tierMasterAgent -> tierMasterAgent.releasePartition(partitionId));
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java
new file mode 100644
index 00000000000..34a0f03f1e1
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java
@@ -0,0 +1,125 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.storage;
+
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.BufferCompressor;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.util.ExceptionUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+/** Client of the Tiered Storage used by the producer. */
+public class TieredStorageProducerClient {
+ private final boolean isBroadcastOnly;
+
+ private final int numSubpartitions;
+
+ private final BufferAccumulator bufferAccumulator;
+
+ private final BufferCompressor bufferCompressor;
+
+ private final List<TierProducerAgent> tierProducerAgents;
+
+ public TieredStorageProducerClient(
+ int numSubpartitions,
+ boolean isBroadcastOnly,
+ BufferAccumulator bufferAccumulator,
+ @Nullable BufferCompressor bufferCompressor,
+ List<TierProducerAgent> tierProducerAgents) {
+ this.isBroadcastOnly = isBroadcastOnly;
+ this.numSubpartitions = numSubpartitions;
+ this.bufferAccumulator = bufferAccumulator;
+ this.bufferCompressor = bufferCompressor;
+ this.tierProducerAgents = tierProducerAgents;
+
+ bufferAccumulator.setup(numSubpartitions, this::writeAccumulatedBuffers);
+ }
+
+ /**
+ * Write records to the producer client. The {@link BufferAccumulator} will accumulate the
+ * records into buffers.
+ *
+ * <p>Note that isBroadcast indicates whether the record is broadcast, while isBroadcastOnly
+ * indicates whether the result partition is broadcast-only. When the result partition is not
+ * broadcast-only and the record is a broadcast record, the record will be written to all the
+ * subpartitions.
+ *
+ * @param record the written record data
+ * @param subpartitionId the subpartition identifier
+ * @param dataType the data type of the record
+ * @param isBroadcast whether the record is a broadcast record
+ */
+ public void write(
+ ByteBuffer record,
+ TieredStorageSubpartitionId subpartitionId,
+ Buffer.DataType dataType,
+ boolean isBroadcast)
+ throws IOException {
+
+ if (isBroadcast && !isBroadcastOnly) {
+ for (int i = 0; i < numSubpartitions; ++i) {
+ bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ }
+ } else {
+ bufferAccumulator.receive(record, subpartitionId, dataType);
+ }
+ }
+
+ public void close() {
+ bufferAccumulator.close();
+ tierProducerAgents.forEach(TierProducerAgent::close);
+ }
+
+ /**
+ * Write the accumulated buffers of this subpartitionId to the appropriate tiers.
+ *
+ * @param subpartitionId the subpartition identifier
+ * @param accumulatedBuffers the accumulated buffers of this subpartition
+ */
+ private void writeAccumulatedBuffers(
+ TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
+ try {
+ for (Buffer finishedBuffer : accumulatedBuffers) {
+ writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ }
+ } catch (IOException e) {
+ ExceptionUtils.rethrow(e);
+ }
+ }
+
+ /**
+ * Write the accumulated buffer of this subpartitionId to an appropriate tier. After the tier is
+ * decided, the buffer will be written to the selected tier.
+ *
+ * @param subpartitionId the subpartition identifier
+ * @param accumulatedBuffer one accumulated buffer of this subpartition
+ */
+ private void writeAccumulatedBuffer(
+ TieredStorageSubpartitionId subpartitionId, Buffer accumulatedBuffer)
+ throws IOException {
+ // TODO, Try to write the accumulated buffer to the appropriate tier. After the tier is
+ // decided, then write the accumulated buffer to the tier.
+ }
+}
[flink] 02/06: [FLINK-31635][network] Introduce the tiered storage configuration
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 14707742d241eadf6d43bbad503db9e9898a8fea
Author: Yuxin Tan <ta...@gmail.com>
AuthorDate: Mon May 8 13:48:30 2023 +0800
[FLINK-31635][network] Introduce the tiered storage configuration
---
.../tiered/common/TieredStorageConfiguration.java | 42 ++++++++++++++++++++++
1 file changed, 42 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
new file mode 100644
index 00000000000..185c267bd6f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/common/TieredStorageConfiguration.java
@@ -0,0 +1,42 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.common;
+
+import org.apache.flink.configuration.Configuration;
+
+/** Configurations for the Tiered Storage. */
+public class TieredStorageConfiguration {
+
+ public static TieredStorageConfiguration.Builder builder() {
+ return new TieredStorageConfiguration.Builder();
+ }
+
+ /** Builder for {@link TieredStorageConfiguration}. */
+ public static class Builder {
+ public TieredStorageConfiguration build() {
+ return new TieredStorageConfiguration();
+ }
+ }
+
+ public static TieredStorageConfiguration fromConfiguration(Configuration conf) {
+ // TODO, from the configuration, get the configured options(i.e., remote storage path, the
+ // reserved storage size, etc.), then set them to the builder.
+ return new TieredStorageConfiguration.Builder().build();
+ }
+}
[flink] 06/06: [FLINK-31635][network] Introduce tiered result partition
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 80a924309ce910715c4079c7e52e9d560318bd38
Author: Yuxin Tan <ta...@gmail.com>
AuthorDate: Mon May 8 13:54:07 2023 +0800
[FLINK-31635][network] Introduce tiered result partition
This closes #22330
---
.../shuffle/TieredInternalShuffleMaster.java | 97 +++++++++
.../tiered/shuffle/TieredResultPartition.java | 225 +++++++++++++++++++++
.../flink/runtime/shuffle/NettyShuffleMaster.java | 15 +-
.../hybrid/tiered/TestingBufferAccumulator.java | 45 +++++
.../tiered/shuffle/TieredResultPartitionTest.java | 216 ++++++++++++++++++++
5 files changed, 597 insertions(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredInternalShuffleMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredInternalShuffleMaster.java
new file mode 100644
index 00000000000..5f8baa3ac36
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredInternalShuffleMaster.java
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.shuffle;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageConfiguration;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageMasterClient;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageResourceRegistry;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierMasterAgent;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageIdMappingUtils.convertId;
+
+/**
+ * A wrapper internal shuffle master class for tiered storage. All the tiered storage operations
+ * with the shuffle master should be wrapped in this class.
+ */
+public class TieredInternalShuffleMaster {
+
+ private final TieredStorageMasterClient tieredStorageMasterClient;
+
+ private final Map<JobID, List<ResultPartitionID>> jobPartitionIds;
+
+ private final Map<ResultPartitionID, JobID> partitionJobIds;
+
+ public TieredInternalShuffleMaster(Configuration conf) {
+ TieredStorageConfiguration tieredStorageConfiguration =
+ TieredStorageConfiguration.fromConfiguration(conf);
+ TieredStorageResourceRegistry resourceRegistry = new TieredStorageResourceRegistry();
+ List<TierMasterAgent> tierFactories =
+ tieredStorageConfiguration.getTierFactories().stream()
+ .map(tierFactory -> tierFactory.createMasterAgent(resourceRegistry))
+ .collect(Collectors.toList());
+ this.tieredStorageMasterClient = new TieredStorageMasterClient(tierFactories);
+ this.jobPartitionIds = new HashMap<>();
+ this.partitionJobIds = new HashMap<>();
+ }
+
+ public void addPartition(JobID jobID, ResultPartitionID resultPartitionID) {
+ jobPartitionIds.computeIfAbsent(jobID, ignore -> new ArrayList<>()).add(resultPartitionID);
+ partitionJobIds.put(resultPartitionID, jobID);
+ tieredStorageMasterClient.addPartition(convertId(resultPartitionID));
+ }
+
+ public void releasePartition(ResultPartitionID resultPartitionID) {
+ tieredStorageMasterClient.releasePartition(convertId(resultPartitionID));
+ JobID jobID = partitionJobIds.remove(resultPartitionID);
+ if (jobID == null) {
+ return;
+ }
+
+ List<ResultPartitionID> resultPartitionIDs = jobPartitionIds.get(jobID);
+ if (resultPartitionIDs == null) {
+ return;
+ }
+
+ resultPartitionIDs.remove(resultPartitionID);
+ // If the result partition id list has been empty, remove the jobID from the map eagerly
+ if (resultPartitionIDs.isEmpty()) {
+ jobPartitionIds.remove(jobID);
+ }
+ }
+
+ public void unregisterJob(JobID jobID) {
+ List<ResultPartitionID> resultPartitionIDs = jobPartitionIds.remove(jobID);
+ if (resultPartitionIDs != null) {
+ resultPartitionIDs.forEach(
+ resultPartitionID -> {
+ tieredStorageMasterClient.releasePartition(convertId(resultPartitionID));
+ partitionJobIds.remove(resultPartitionID);
+ });
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartition.java
new file mode 100644
index 00000000000..cd24ac89577
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartition.java
@@ -0,0 +1,225 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.shuffle;
+
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.event.AbstractEvent;
+import org.apache.flink.runtime.io.network.api.EndOfData;
+import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
+import org.apache.flink.runtime.io.network.api.StopMode;
+import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.BufferCompressor;
+import org.apache.flink.runtime.io.network.buffer.BufferPool;
+import org.apache.flink.runtime.io.network.partition.BufferAvailabilityListener;
+import org.apache.flink.runtime.io.network.partition.ResultPartition;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageIdMappingUtils;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageProducerClient;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageResourceRegistry;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.util.concurrent.FutureUtils;
+import org.apache.flink.util.function.SupplierWithException;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * {@link TieredResultPartition} appends records and events to the tiered storage, which supports
+ * the upstream dynamically switches storage tier for writing shuffle data, and the downstream will
+ * read data from the relevant tier.
+ */
+public class TieredResultPartition extends ResultPartition {
+
+ private final TieredStoragePartitionId partitionId;
+
+ private final TieredStorageProducerClient tieredStorageProducerClient;
+
+ private final TieredStorageResourceRegistry tieredStorageResourceRegistry;
+
+ private boolean hasNotifiedEndOfUserRecords;
+
+ public TieredResultPartition(
+ String owningTaskName,
+ int partitionIndex,
+ ResultPartitionID partitionId,
+ ResultPartitionType partitionType,
+ int numSubpartitions,
+ int numTargetKeyGroups,
+ ResultPartitionManager partitionManager,
+ @Nullable BufferCompressor bufferCompressor,
+ SupplierWithException<BufferPool, IOException> bufferPoolFactory,
+ TieredStorageProducerClient tieredStorageProducerClient,
+ TieredStorageResourceRegistry tieredStorageResourceRegistry) {
+ super(
+ owningTaskName,
+ partitionIndex,
+ partitionId,
+ partitionType,
+ numSubpartitions,
+ numTargetKeyGroups,
+ partitionManager,
+ bufferCompressor,
+ bufferPoolFactory);
+
+ this.partitionId = TieredStorageIdMappingUtils.convertId(partitionId);
+ this.tieredStorageProducerClient = tieredStorageProducerClient;
+ this.tieredStorageResourceRegistry = tieredStorageResourceRegistry;
+ }
+
+ @Override
+ protected void setupInternal() throws IOException {
+ if (isReleased()) {
+ throw new IOException("Result partition has been released.");
+ }
+ }
+
+ @Override
+ public void setMetricGroup(TaskIOMetricGroup metrics) {
+ super.setMetricGroup(metrics);
+ }
+
+ @Override
+ public void emitRecord(ByteBuffer record, int consumerId) throws IOException {
+ resultPartitionBytes.inc(consumerId, record.remaining());
+ emit(record, consumerId, Buffer.DataType.DATA_BUFFER, false);
+ }
+
+ @Override
+ public void broadcastRecord(ByteBuffer record) throws IOException {
+ resultPartitionBytes.incAll(record.remaining());
+ broadcast(record, Buffer.DataType.DATA_BUFFER);
+ }
+
+ @Override
+ public void broadcastEvent(AbstractEvent event, boolean isPriorityEvent) throws IOException {
+ Buffer buffer = EventSerializer.toBuffer(event, isPriorityEvent);
+ try {
+ ByteBuffer serializedEvent = buffer.getNioBufferReadable();
+ broadcast(serializedEvent, buffer.getDataType());
+ } finally {
+ buffer.recycleBuffer();
+ }
+ }
+
+ private void broadcast(ByteBuffer record, Buffer.DataType dataType) throws IOException {
+ checkInProduceState();
+ emit(record, 0, dataType, true);
+ }
+
+ private void emit(
+ ByteBuffer record, int consumerId, Buffer.DataType dataType, boolean isBroadcast)
+ throws IOException {
+ tieredStorageProducerClient.write(
+ record, TieredStorageIdMappingUtils.convertId(consumerId), dataType, isBroadcast);
+ }
+
+ @Override
+ public ResultSubpartitionView createSubpartitionView(
+ int subpartitionId, BufferAvailabilityListener availabilityListener)
+ throws IOException {
+ checkState(!isReleased(), "ResultPartition already released.");
+ // TODO, create subpartition views
+ return null;
+ }
+
+ @Override
+ public void finish() throws IOException {
+ broadcastEvent(EndOfPartitionEvent.INSTANCE, false);
+ checkState(!isReleased(), "Result partition is already released.");
+ super.finish();
+ }
+
+ @Override
+ public void close() {
+ super.close();
+ tieredStorageProducerClient.close();
+ }
+
+ @Override
+ protected void releaseInternal() {
+ tieredStorageResourceRegistry.clearResourceFor(partitionId);
+ }
+
+ @Override
+ public void notifyEndOfData(StopMode mode) throws IOException {
+ if (!hasNotifiedEndOfUserRecords) {
+ broadcastEvent(new EndOfData(mode), false);
+ hasNotifiedEndOfUserRecords = true;
+ }
+ }
+
+ @Override
+ public void alignedBarrierTimeout(long checkpointId) throws IOException {
+ // Nothing to do.
+ }
+
+ @Override
+ public void abortCheckpoint(long checkpointId, CheckpointException cause) {
+ // Nothing to do.
+ }
+
+ @Override
+ public void flushAll() {
+ // Nothing to do.
+ }
+
+ @Override
+ public void flush(int subpartitionIndex) {
+ // Nothing to do.
+ }
+
+ @Override
+ public CompletableFuture<Void> getAllDataProcessedFuture() {
+ // Nothing to do.
+ return FutureUtils.completedVoidFuture();
+ }
+
+ @Override
+ public void onSubpartitionAllDataProcessed(int subpartition) {
+ // Nothing to do.
+ }
+
+ @Override
+ public int getNumberOfQueuedBuffers() {
+ // Nothing to do.
+ return 0;
+ }
+
+ @Override
+ public long getSizeOfQueuedBuffersUnsafe() {
+ // Nothing to do.
+ return 0;
+ }
+
+ @Override
+ public int getNumberOfQueuedBuffers(int targetSubpartition) {
+ // Nothing to do.
+ return 0;
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
index 7beb385d395..75441b7331f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.shuffle.TieredInternalShuffleMaster;
import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor.LocalExecutionPartitionConnectionInfo;
import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor.NetworkPartitionConnectionInfo;
import org.apache.flink.runtime.shuffle.NettyShuffleDescriptor.PartitionConnectionInfo;
@@ -49,6 +50,8 @@ public class NettyShuffleMaster implements ShuffleMaster<NettyShuffleDescriptor>
private final int networkBufferSize;
+ private final TieredInternalShuffleMaster tieredInternalShuffleMaster;
+
public NettyShuffleMaster(Configuration conf) {
checkNotNull(conf);
buffersPerInputChannel =
@@ -64,6 +67,7 @@ public class NettyShuffleMaster implements ShuffleMaster<NettyShuffleDescriptor>
sortShuffleMinBuffers =
conf.getInteger(NettyShuffleEnvironmentOptions.NETWORK_SORT_SHUFFLE_MIN_BUFFERS);
networkBufferSize = ConfigurationParserUtils.getPageSize(conf);
+ tieredInternalShuffleMaster = new TieredInternalShuffleMaster(conf);
checkArgument(
!maxRequiredBuffersPerGate.isPresent() || maxRequiredBuffersPerGate.get() >= 1,
@@ -96,11 +100,20 @@ public class NettyShuffleMaster implements ShuffleMaster<NettyShuffleDescriptor>
producerDescriptor, partitionDescriptor.getConnectionIndex()),
resultPartitionID);
+ tieredInternalShuffleMaster.addPartition(jobID, resultPartitionID);
+
return CompletableFuture.completedFuture(shuffleDeploymentDescriptor);
}
@Override
- public void releasePartitionExternally(ShuffleDescriptor shuffleDescriptor) {}
+ public void releasePartitionExternally(ShuffleDescriptor shuffleDescriptor) {
+ tieredInternalShuffleMaster.releasePartition(shuffleDescriptor.getResultPartitionID());
+ }
+
+ @Override
+ public void unregisterJob(JobID jobID) {
+ tieredInternalShuffleMaster.unregisterJob(jobID);
+ }
private static PartitionConnectionInfo createConnectionInfo(
ProducerDescriptor producerDescriptor, int connectionIndex) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingBufferAccumulator.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingBufferAccumulator.java
new file mode 100644
index 00000000000..9430f5f3b2e
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingBufferAccumulator.java
@@ -0,0 +1,45 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered;
+
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.BufferAccumulator;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.function.BiConsumer;
+
+/** Test implementation for {@link BufferAccumulator}. */
+public class TestingBufferAccumulator implements BufferAccumulator {
+
+ @Override
+ public void setup(
+ int numSubpartitions,
+ BiConsumer<TieredStorageSubpartitionId, List<Buffer>> bufferFlusher) {}
+
+ @Override
+ public void receive(
+ ByteBuffer record, TieredStorageSubpartitionId subpartitionId, Buffer.DataType dataType)
+ throws IOException {}
+
+ @Override
+ public void close() {}
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartitionTest.java
new file mode 100644
index 00000000000..7e69f015366
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/shuffle/TieredResultPartitionTest.java
@@ -0,0 +1,216 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.shuffle;
+
+import org.apache.flink.runtime.executiongraph.IOMetrics;
+import org.apache.flink.runtime.executiongraph.ResultPartitionBytes;
+import org.apache.flink.runtime.io.disk.BatchShuffleReadBufferPool;
+import org.apache.flink.runtime.io.disk.FileChannelManager;
+import org.apache.flink.runtime.io.disk.FileChannelManagerImpl;
+import org.apache.flink.runtime.io.network.buffer.BufferCompressor;
+import org.apache.flink.runtime.io.network.buffer.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.NoOpBufferAvailablityListener;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageProducerClient;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageResourceRegistry;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredResultPartition}. */
+class TieredResultPartitionTest {
+
+ private static final int NUM_THREADS = 4;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NUM_TOTAL_BYTES_IN_READ_POOL = 32 * 1024 * 1024;
+
+ private FileChannelManager fileChannelManager;
+
+ private NetworkBufferPool globalPool;
+
+ private BatchShuffleReadBufferPool readBufferPool;
+
+ private ScheduledExecutorService readIOExecutor;
+
+ private TaskIOMetricGroup taskIOMetricGroup;
+
+ @TempDir public java.nio.file.Path tempDataPath;
+
+ @BeforeEach
+ void before() {
+ fileChannelManager =
+ new FileChannelManagerImpl(new String[] {tempDataPath.toString()}, "testing");
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ readBufferPool =
+ new BatchShuffleReadBufferPool(NUM_TOTAL_BYTES_IN_READ_POOL, NETWORK_BUFFER_SIZE);
+ readIOExecutor =
+ new ScheduledThreadPoolExecutor(
+ NUM_THREADS,
+ new ExecutorThreadFactory("test-io-scheduler-thread"),
+ (ignored, executor) -> {
+ if (executor.isShutdown()) {
+ // ignore rejected as shutdown.
+ } else {
+ throw new RejectedExecutionException();
+ }
+ });
+ }
+
+ @AfterEach
+ void after() throws Exception {
+ fileChannelManager.close();
+ globalPool.destroy();
+ readBufferPool.destroy();
+ readIOExecutor.shutdown();
+ }
+
+ @Test
+ void testClose() throws Exception {
+ final int numBuffers = 1;
+
+ BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
+ TieredResultPartition partition = createTieredStoreResultPartition(1, bufferPool, false);
+
+ partition.close();
+ assertThat(bufferPool.isDestroyed()).isTrue();
+ }
+
+ @Test
+ @Timeout(30)
+ void testRelease() throws Exception {
+ final int numSubpartitions = 2;
+ final int numBuffers = 10;
+
+ BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
+ TieredResultPartition partition =
+ createTieredStoreResultPartition(numSubpartitions, bufferPool, false);
+
+ partition.emitRecord(ByteBuffer.allocate(NETWORK_BUFFER_SIZE * 5), 1);
+ partition.close();
+ assertThat(bufferPool.isDestroyed()).isTrue();
+
+ partition.release();
+
+ while (checkNotNull(fileChannelManager.getPaths()[0].listFiles()).length != 0) {
+ Thread.sleep(10);
+ }
+
+ assertThat(NUM_TOTAL_BUFFERS).isEqualTo(globalPool.getNumberOfAvailableMemorySegments());
+ }
+
+ @Test
+ void testCreateSubpartitionViewAfterRelease() throws Exception {
+ final int numBuffers = 10;
+ BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
+ TieredResultPartition resultPartition =
+ createTieredStoreResultPartition(2, bufferPool, false);
+ resultPartition.release();
+ assertThatThrownBy(
+ () ->
+ resultPartition.createSubpartitionView(
+ 0, new NoOpBufferAvailablityListener()))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void testEmitRecords() throws Exception {
+ BufferPool bufferPool = globalPool.createBufferPool(3, 3);
+ int bufferSize = NETWORK_BUFFER_SIZE;
+ try (TieredResultPartition partition =
+ createTieredStoreResultPartition(2, bufferPool, false)) {
+ partition.emitRecord(ByteBuffer.allocate(bufferSize), 0);
+ partition.broadcastRecord(ByteBuffer.allocate(bufferSize));
+ IOMetrics ioMetrics = taskIOMetricGroup.createSnapshot();
+ assertThat(ioMetrics.getResultPartitionBytes()).hasSize(1);
+ ResultPartitionBytes partitionBytes =
+ ioMetrics.getResultPartitionBytes().values().iterator().next();
+ assertThat(partitionBytes.getSubpartitionBytes())
+ .containsExactly((long) 2 * bufferSize, bufferSize);
+ }
+ }
+
+ @Test
+ void testMetricsUpdateForBroadcastOnlyResultPartition() throws Exception {
+ BufferPool bufferPool = globalPool.createBufferPool(3, 3);
+ int bufferSize = NETWORK_BUFFER_SIZE;
+ try (TieredResultPartition partition =
+ createTieredStoreResultPartition(2, bufferPool, true)) {
+ partition.broadcastRecord(ByteBuffer.allocate(bufferSize));
+ IOMetrics ioMetrics = taskIOMetricGroup.createSnapshot();
+ assertThat(ioMetrics.getResultPartitionBytes()).hasSize(1);
+ ResultPartitionBytes partitionBytes =
+ ioMetrics.getResultPartitionBytes().values().iterator().next();
+ assertThat(partitionBytes.getSubpartitionBytes())
+ .containsExactly(bufferSize, bufferSize);
+ }
+ }
+
+ private TieredResultPartition createTieredStoreResultPartition(
+ int numSubpartitions, BufferPool bufferPool, boolean isBroadcastOnly)
+ throws IOException {
+ TieredResultPartition tieredResultPartition =
+ new TieredResultPartition(
+ "TieredStoreResultPartitionTest",
+ 0,
+ new ResultPartitionID(),
+ ResultPartitionType.HYBRID_SELECTIVE,
+ numSubpartitions,
+ numSubpartitions,
+ new ResultPartitionManager(),
+ new BufferCompressor(NETWORK_BUFFER_SIZE, "LZ4"),
+ () -> bufferPool,
+ new TieredStorageProducerClient(
+ numSubpartitions,
+ isBroadcastOnly,
+ new TestingBufferAccumulator(),
+ null,
+ new ArrayList<>()),
+ new TieredStorageResourceRegistry());
+ taskIOMetricGroup =
+ UnregisteredMetricGroups.createUnregisteredTaskMetricGroup().getIOMetricGroup();
+ tieredResultPartition.setup();
+ tieredResultPartition.setMetricGroup(taskIOMetricGroup);
+ return tieredResultPartition;
+ }
+}
[flink] 04/06: [FLINK-31635][network] Introduce tiered storage resource and the resource registry
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bd11c4036eff2e6cefcaabef0658e4f5886e109b
Author: Yuxin Tan <ta...@gmail.com>
AuthorDate: Wed May 10 19:49:01 2023 +0800
[FLINK-31635][network] Introduce tiered storage resource and the resource registry
---
.../TieredStorageResource.java} | 16 ++----
.../storage/TieredStorageResourceRegistry.java | 62 ++++++++++++++++++++++
.../partition/hybrid/tiered/tier/TierFactory.java | 4 +-
3 files changed, 70 insertions(+), 12 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageResource.java
similarity index 68%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageResource.java
index 3225c3a5a09..fa1d85111ed 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageResource.java
@@ -16,17 +16,11 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier;
+package org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage;
-/** A factory that creates all the components of a tier. */
-public interface TierFactory {
+/** The resource (e.g., local files, remote storage files, etc.) for the Tiered Storage. */
+public interface TieredStorageResource {
- /** Creates the master-side agent of a Tier. */
- TierMasterAgent createMasterAgent();
-
- /** Creates the producer-side agent of a Tier. */
- TierProducerAgent createProducerAgent();
-
- /** Creates the consumer-side agent of a Tier. */
- TierConsumerAgent createConsumerAgent();
+ /** Release all the resources, e.g. delete the files, recycle the occupied memory, etc. */
+ void release();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageResourceRegistry.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageResourceRegistry.java
new file mode 100644
index 00000000000..cae47f969f7
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageResourceRegistry.java
@@ -0,0 +1,62 @@
+/*
+ * 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.runtime.io.network.partition.hybrid.tiered.storage;
+
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageDataIdentifier;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A registry that maintains local or remote resources that correspond to a certain set of data in
+ * the Tiered Storage.
+ */
+public class TieredStorageResourceRegistry {
+
+ private final Map<TieredStorageDataIdentifier, List<TieredStorageResource>>
+ registeredResources = new HashMap<>();
+
+ /**
+ * Register a new resource for the given owner.
+ *
+ * @param owner identifier of the data that the resource corresponds to.
+ * @param tieredStorageResource the tiered storage resources to be registered.
+ */
+ public void registerResource(
+ TieredStorageDataIdentifier owner, TieredStorageResource tieredStorageResource) {
+ registeredResources
+ .computeIfAbsent(owner, (ignore) -> new ArrayList<>())
+ .add(tieredStorageResource);
+ }
+
+ /**
+ * Remove all resources for the given owner.
+ *
+ * @param owner identifier of the data that the resources correspond to.
+ */
+ public void clearResourceFor(TieredStorageDataIdentifier owner) {
+ List<TieredStorageResource> cleanersForOwner = registeredResources.remove(owner);
+
+ if (cleanersForOwner != null) {
+ cleanersForOwner.forEach(TieredStorageResource::release);
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
index 3225c3a5a09..5499fc59ff6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierFactory.java
@@ -18,11 +18,13 @@
package org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.TieredStorageResourceRegistry;
+
/** A factory that creates all the components of a tier. */
public interface TierFactory {
/** Creates the master-side agent of a Tier. */
- TierMasterAgent createMasterAgent();
+ TierMasterAgent createMasterAgent(TieredStorageResourceRegistry tieredStorageResourceRegistry);
/** Creates the producer-side agent of a Tier. */
TierProducerAgent createProducerAgent();