You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "TanYuxin-tyx (via GitHub)" <gi...@apache.org> on 2023/05/25 07:13:50 UTC
[GitHub] [flink] TanYuxin-tyx opened a new pull request, #22652: [FLINK-31640][network] Introduce the segment id tracker for the tiered storage
TanYuxin-tyx opened a new pull request, #22652:
URL: https://github.com/apache/flink/pull/22652
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
1. Introduce the segment id tracker for the tiered storage
2. Write the accumulated buffers to the right storage tier
## Brief change log
- ***[Commit 1]**: Introduce the segment id tracker and its implementation for the tiered storage*
- ***[Commit 2]**: Firstly choose the storage tier, then write the accumulated buffers to the right storage tier.*
## Verifying this change
- *Added the test of `SubpartitionSegmentIdTrackerImplTest` to verify the logic of `SubpartitionSegmentIdTrackerImpl`*
- *Added the test of `TieredStorageProducerClientTest` to verify the logic of choosing storage tiers.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on PR #22652:
URL: https://github.com/apache/flink/pull/22652#issuecomment-1575978610
I rebased on the master branch to trigger tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211220492
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTrackerImpl.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.util.function.SupplierWithException;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * The implementation of {@link SubpartitionSegmentIdTracker}. Each {@link TierProducerAgent} has a
+ * {@link SubpartitionSegmentIdTrackerImpl} to record the segments belong to the producer agent.
+ */
+public class SubpartitionSegmentIdTrackerImpl implements SubpartitionSegmentIdTracker {
+
+ /**
+ * Each subpartition calculates the amount of data written to a tier separately. If the amount
+ * of data exceeds the threshold, the current writing segment will be finished. Because
+ * different subpartitions may have duplicate segment indexes, this field needs to distinguish
+ * between different subpartitions.
+ */
+ private final Map<TieredStorageSubpartitionId, HashSet<Integer>> subpartitionSegmentIds;
+
+ /** Indicate whether this is a broadcast only partition. */
+ private final Boolean isBroadCastOnly;
+
+ /** Record the latest segment index for each subpartition. */
+ private final int[] latestSegmentIds;
+
+ /** The locks for all subpartitions. */
+ private final Lock[] locks;
+
+ public SubpartitionSegmentIdTrackerImpl(int numSubpartitions, Boolean isBroadcastOnly) {
+ this.isBroadCastOnly = isBroadcastOnly;
+ int effectiveNumSubpartitions = isBroadcastOnly ? 1 : numSubpartitions;
+ this.latestSegmentIds = new int[effectiveNumSubpartitions];
+ this.subpartitionSegmentIds = new HashMap<>();
+ this.locks = new Lock[effectiveNumSubpartitions];
+
+ Arrays.fill(latestSegmentIds, -1);
+ for (int i = 0; i < effectiveNumSubpartitions; i++) {
+ locks[i] = new ReentrantLock();
+ subpartitionSegmentIds.put(new TieredStorageSubpartitionId(i), new HashSet<>());
Review Comment:
Based on the offline discussion, the segment tracker will be introduced in the subsequential PR.
When introducing the segment tracker, this will be fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212861475
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,26 +137,95 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
- try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ Iterator<Buffer> bufferIterator = accumulatedBuffers.iterator();
+
+ int numWriteBytes = 0;
+ int numWriteBuffers = 0;
+ while (bufferIterator.hasNext()) {
+ Buffer buffer = bufferIterator.next();
Review Comment:
What we need to calculate should be the size before compression in order to align with the downstream `numBytesRead`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211225894
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +145,66 @@ private void writeAccumulatedBuffers(
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.
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int tierIndex = chooseStorageTierIfNeeded(subpartitionId);
+
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+ updateStatistics(compressedBuffer);
+ boolean isLastBufferInSegment =
+ tierProducerAgents.get(tierIndex).write(subpartitionId, compressedBuffer);
+ if (isLastBufferInSegment) {
+ isSubpartitionSegmentFinished[subpartitionIndex] = true;
+ }
+ }
+
+ private int chooseStorageTierIfNeeded(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ if (isSubpartitionSegmentFinished[subpartitionIndex]) {
+ // If the current segment is finished, this subpartition need choose a new storage tier.
+ internalChooseStorageTier(subpartitionId);
+ isSubpartitionSegmentFinished[subpartitionIndex] = false;
+ }
+ return currentSubpartitionTierIndex[subpartitionIndex];
+ }
+
+ private void internalChooseStorageTier(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (int tierIndex = 0; tierIndex < tierProducerAgents.size(); ++tierIndex) {
+ if (tierProducerAgents
+ .get(tierIndex)
Review Comment:
Fixed. Replace it by `currentSubpartitionTierAgent`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211228446
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +145,66 @@ private void writeAccumulatedBuffers(
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.
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int tierIndex = chooseStorageTierIfNeeded(subpartitionId);
+
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+ updateStatistics(compressedBuffer);
+ boolean isLastBufferInSegment =
+ tierProducerAgents.get(tierIndex).write(subpartitionId, compressedBuffer);
Review Comment:
OK, I have changed the logic of choosing a new tier.
The return value of `write` show whether the record can be written to the tier. If true, continue writing. If not, it will choose a new storage tier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214223698
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
+
+ private Supplier<Boolean[]> tryWriteReturnValueSupplier;
Review Comment:
Fixed.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
+
+ private Supplier<Boolean[]> tryWriteReturnValueSupplier;
+
+ private int numTotalReceivedBuffers;
+
+ private int tryWriteCounter;
+
+ private boolean isClosed;
+
+ public void setTryStartNewSegmentReturnValueSupplier(
+ Supplier<Boolean> tryStartNewSegmentReturnValueSupplier) {
+ this.tryStartNewSegmentReturnValueSupplier = tryStartNewSegmentReturnValueSupplier;
+ }
+
+ public void setTryWriteReturnValueSupplier(Supplier<Boolean[]> tryWriteReturnValueSupplier) {
+ this.tryWriteReturnValueSupplier = tryWriteReturnValueSupplier;
+ }
+
+ public int numTotalReceivedBuffers() {
+ return numTotalReceivedBuffers;
+ }
+
+ public boolean isClosed() {
+ return isClosed;
+ }
+
+ @Override
+ public boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId) {
+ return tryStartNewSegmentReturnValueSupplier.get();
+ }
+
+ @Override
+ public boolean tryWrite(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer) {
+ boolean isSuccess = tryWriteReturnValueSupplier.get()[(tryWriteCounter++) % 2];
+ if (isSuccess) {
+ numTotalReceivedBuffers++;
+ }
+ return isSuccess;
+ }
+
+ @Override
+ public void close() {
+ this.isClosed = true;
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on PR #22652:
URL: https://github.com/apache/flink/pull/22652#issuecomment-1569708841
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212795159
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,26 +137,95 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
- try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ Iterator<Buffer> bufferIterator = accumulatedBuffers.iterator();
+
+ int numWriteBytes = 0;
+ int numWriteBuffers = 0;
+ while (bufferIterator.hasNext()) {
+ Buffer buffer = bufferIterator.next();
+ try {
+ writeAccumulatedBuffer(subpartitionId, buffer);
+ } catch (IOException ioe) {
+ buffer.recycleBuffer();
+ while (bufferIterator.hasNext()) {
+ bufferIterator.next().recycleBuffer();
+ }
+ ExceptionUtils.rethrow(ioe);
}
- } catch (IOException e) {
- ExceptionUtils.rethrow(e);
+ numWriteBuffers++;
+ numWriteBytes += buffer.readableBytes();
}
+ updateMetricStatistics(numWriteBuffers, numWriteBytes);
}
/**
* 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.
*
+ * <p>Note that the method only throws an exception when choosing a storage tier, so the caller
+ * should ensure that the buffer is recycled when throwing an exception.
+ *
* @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.
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ if (!isSuccess) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ checkState(isSuccess, "Failed to write the first buffer to the new segment");
+ }
+ }
+
+ private void chooseStorageTierToStartSegment(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (TierProducerAgent tierProducerAgent : tierProducerAgents) {
+ if (tierProducerAgent.tryStartNewSegment(subpartitionId, nextSegmentIndex)) {
+ // Update the segment index and the chosen storage tier for the subpartition.
+ currentSubpartitionSegmentId[subpartitionIndex] = nextSegmentIndex;
+ currentSubpartitionTierAgent[subpartitionIndex] = tierProducerAgent;
+ return;
+ }
+ }
+ throw new IOException("Failed to choose a storage tier to start a new segment.");
+ }
+
+ private Buffer compressBufferIfPossible(Buffer buffer) {
+ if (!canBeCompressed(buffer)) {
+ return buffer;
+ }
+
+ return checkNotNull(bufferCompressor).compressToOriginalBuffer(buffer);
+ }
+
+ /**
+ * Whether the buffer can be compressed or not. Note that event is not compressed because it is
+ * usually small and the size can become even larger after compression.
+ */
+ private boolean canBeCompressed(Buffer buffer) {
+ return bufferCompressor != null && buffer.isBuffer() && buffer.readableBytes() > 0;
+ }
+
+ private void updateMetricStatistics(int numWriteBuffers, int numWriteBytes) {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214031770
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java:
##########
@@ -35,9 +41,21 @@ public interface TierProducerAgent {
*/
boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId);
- /** Writes the finished {@link Buffer} to the consumer. */
- boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer)
- throws IOException;
+ /**
+ * Writes the finished {@link Buffer} to the consumer.
+ *
+ * <p>Note that the tier must ensure that the buffer is written successfully without any
+ * exceptions, in order to guarantee that the buffer will be recycled. If this method throws an
+ * exception in the subsequent modifications, the caller should make sure that the buffer is
+ * recycled finally.
+ *
+ * @param subpartitionId the subpartition id that the buffer is writing to
+ * @param finishedBuffer the writing buffer
+ * @return return true if the buffer is written successfully, return false if the current
+ * segment can not store this buffer and the current segment is finished. When returning
+ * false, the agent should try start a new segment before writing the buffer.
+ */
+ boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer);
Review Comment:
Renamed it.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,26 +137,95 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
- try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ Iterator<Buffer> bufferIterator = accumulatedBuffers.iterator();
+
+ int numWriteBytes = 0;
+ int numWriteBuffers = 0;
+ while (bufferIterator.hasNext()) {
+ Buffer buffer = bufferIterator.next();
+ try {
+ writeAccumulatedBuffer(subpartitionId, buffer);
+ } catch (IOException ioe) {
+ buffer.recycleBuffer();
+ while (bufferIterator.hasNext()) {
+ bufferIterator.next().recycleBuffer();
+ }
+ ExceptionUtils.rethrow(ioe);
}
- } catch (IOException e) {
- ExceptionUtils.rethrow(e);
+ numWriteBuffers++;
+ numWriteBytes += buffer.readableBytes();
}
+ updateMetricStatistics(numWriteBuffers, numWriteBytes);
}
/**
* 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.
*
+ * <p>Note that the method only throws an exception when choosing a storage tier, so the caller
+ * should ensure that the buffer is recycled when throwing an exception.
+ *
* @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.
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ if (!isSuccess) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ checkState(isSuccess, "Failed to write the first buffer to the new segment");
+ }
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214036888
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, Collections.emptyList());
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testEmptyMetricUpdater() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+ TestingTierProducerAgent testingTierProducerAgent =
+ new TestingTierProducerAgent(false, numSubpartitions);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions,
+ false,
+ bufferAccumulator,
+ null,
+ Collections.singletonList(testingTierProducerAgent));
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ @TestTemplate
+ void testWriteRecordsToMultipleStorageTiers() throws IOException {
Review Comment:
Simplify the logic by introducing a `TestingTierProducerAgent`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on PR #22652:
URL: https://github.com/apache/flink/pull/22652#issuecomment-1576310784
Thanks @xintongsong and @reswqa for the careful review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214225246
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingTierProducerAgent;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.TieredStorageTestUtils.generateRandomData;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(numSubpartitions, Collections.emptyList());
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testWriteRecords() throws IOException {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ int maxNumToWriteRecordsPerSubpartition = 1000;
+ Random random = new Random();
+
+ TestingTierProducerAgent tierProducerAgent = new TestingTierProducerAgent();
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> true);
+ tierProducerAgent.setTryWriteReturnValueSupplier(() -> new Boolean[] {false, true});
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(
+ numSubpartitions, Collections.singletonList(tierProducerAgent));
+
+ int numWriteRecords = 0;
+ for (int j = 0; j < numSubpartitions; j++) {
+ for (int i = 0; i < random.nextInt(maxNumToWriteRecordsPerSubpartition); i++) {
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(j),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast);
+ numWriteRecords++;
+ }
+ }
+
+ int numExpectedBuffers = isBroadcast ? numWriteRecords * numSubpartitions : numWriteRecords;
Review Comment:
Fixed. But in order to mock the scenario of writing failed, then choosing a new tier, then writing success, I created two testing tiers.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingTierProducerAgent;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.TieredStorageTestUtils.generateRandomData;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212794310
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerMetricStatistics.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.storage;
+
+/** The metric statistics for the tiered storage producer. */
+public class TieredStorageProducerMetricStatistics {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214037314
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, Collections.emptyList());
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testEmptyMetricUpdater() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+ TestingTierProducerAgent testingTierProducerAgent =
+ new TestingTierProducerAgent(false, numSubpartitions);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions,
+ false,
+ bufferAccumulator,
+ null,
+ Collections.singletonList(testingTierProducerAgent));
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ @TestTemplate
+ void testWriteRecordsToMultipleStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int numSegmentCapacityInLimitedTier = 12;
+ int numMaxToSendRecords = 200;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+ List<TierProducerAgent> tierProducerAgents = new ArrayList<>();
+ TestingTierProducerAgent limitedSegmentsTierProducerAgent =
+ new TestingTierProducerAgent(
+ true, numSubpartitions, numSegmentCapacityInLimitedTier);
+ TestingTierProducerAgent unLimitedTierProducerAgent =
+ new TestingTierProducerAgent(false, numSubpartitions);
+ tierProducerAgents.add(limitedSegmentsTierProducerAgent);
+ tierProducerAgents.add(unLimitedTierProducerAgent);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, tierProducerAgents);
+
+ AtomicInteger numWriteBuffers = new AtomicInteger();
+ AtomicInteger numWriteBytes = new AtomicInteger();
+ tieredStorageProducerClient.setMetricStatisticsUpdater(
+ metricStatistics -> {
+ numWriteBuffers.set(
+ numWriteBuffers.get() + metricStatistics.numWriteBuffersDelta());
+ numWriteBytes.set(numWriteBytes.get() + metricStatistics.numWriteBytesDelta());
+ });
+
+ int[] numSentRecords = new int[numSubpartitions];
+ for (int i = 0; i < numSubpartitions; i++) {
+ int numToSend = random.nextInt(numMaxToSendRecords - 1) + 1;
+ numSentRecords[i] = numToSend;
+ TieredStorageSubpartitionId subpartitionId = new TieredStorageSubpartitionId(i);
+ for (int j = 0; j < numToSend; j++) {
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ subpartitionId,
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast);
+ }
+ }
+
+ checkReceivedBuffersAndSegments(
+ numSubpartitions,
+ bufferSize,
+ numSegmentCapacityInLimitedTier,
+ limitedSegmentsTierProducerAgent,
+ unLimitedTierProducerAgent,
+ numSentRecords,
+ numWriteBuffers,
+ numWriteBytes);
+
+ assertThat(limitedSegmentsTierProducerAgent.isClosed).isFalse();
+ assertThat(unLimitedTierProducerAgent.isClosed).isFalse();
+ tieredStorageProducerClient.close();
+ assertThat(limitedSegmentsTierProducerAgent.isClosed).isTrue();
+ assertThat(unLimitedTierProducerAgent.isClosed).isTrue();
+ storageMemoryManager.release();
+ }
+
+ private void checkReceivedBuffersAndSegments(
+ int numSubpartitions,
+ int bufferSize,
+ int numSegmentCapacityInLimitedTier,
+ TestingTierProducerAgent limitedSegmentsTierProducerAgent,
+ TestingTierProducerAgent unLimitedTierProducerAgent,
+ int[] numSentRecords,
+ AtomicInteger numWriteBuffers,
+ AtomicInteger numWriteBytes) {
+ int[] numExpectedBuffersInLimitedTier = new int[numSubpartitions];
+ int[] numExpectedBuffersInUnLimitedTier = new int[numSubpartitions];
+ int[] numExpectedSegmentInLimitedTier = new int[numSubpartitions];
+ int[] numExpectedSegmentInUnLimitedTier = new int[numSubpartitions];
+
+ calculateNumExpectedBuffersAndSegments(
+ numSubpartitions,
+ numSegmentCapacityInLimitedTier,
+ numSentRecords,
+ numExpectedBuffersInLimitedTier,
+ numExpectedBuffersInUnLimitedTier,
+ numExpectedSegmentInLimitedTier,
+ numExpectedSegmentInUnLimitedTier);
+
+ for (int i = 0; i < numSubpartitions; i++) {
+ assertThat(limitedSegmentsTierProducerAgent.numTotalReceivedBuffers[i])
+ .isEqualTo(numExpectedBuffersInLimitedTier[i]);
+ assertThat(unLimitedTierProducerAgent.numTotalReceivedBuffers[i])
+ .isEqualTo(numExpectedBuffersInUnLimitedTier[i]);
+ assertThat(limitedSegmentsTierProducerAgent.numTotalReceivedBuffers[i] > 0).isTrue();
+ assertThat(limitedSegmentsTierProducerAgent.numReceivedSegments[i])
+ .isEqualTo(numExpectedSegmentInLimitedTier[i]);
+ assertThat(unLimitedTierProducerAgent.numReceivedSegments[i])
+ .isEqualTo(numExpectedSegmentInUnLimitedTier[i]);
+ }
+
+ int numTotalRecords = Arrays.stream(numSentRecords).sum();
+ int numExpectedBuffers = isBroadcast ? numTotalRecords * numSubpartitions : numTotalRecords;
+ assertThat(numWriteBuffers.get()).isEqualTo(numExpectedBuffers);
+ assertThat(numWriteBytes.get()).isEqualTo(numExpectedBuffers * bufferSize);
+ }
+
+ private void calculateNumExpectedBuffersAndSegments(
+ int numSubpartitions,
+ int numSegmentCapacityInLimitedTier,
+ int[] numSentRecords,
+ int[] numExpectedBuffersInLimitedTier,
+ int[] numExpectedBuffersInUnLimitedTier,
+ int[] numExpectedSegmentInLimitedTier,
+ int[] numExpectedSegmentInUnLimitedTier) {
+ int numTotalRecords = Arrays.stream(numSentRecords).map(Integer::new).sum();
+ for (int i = 0; i < numSubpartitions; i++) {
+ int totalSentRecords = isBroadcast ? numTotalRecords : numSentRecords[i];
+ int maxBuffersInLimitedTier =
+ numSegmentCapacityInLimitedTier * NUM_BUFFERS_IN_A_SEGMENT;
+
+ numExpectedBuffersInLimitedTier[i] =
+ Math.min(totalSentRecords, maxBuffersInLimitedTier);
+ numExpectedSegmentInLimitedTier[i] =
+ numExpectedBuffersInLimitedTier[i] / NUM_BUFFERS_IN_A_SEGMENT
+ + (numExpectedBuffersInLimitedTier[i] % NUM_BUFFERS_IN_A_SEGMENT == 0
+ ? 0
+ : 1);
+ numExpectedBuffersInUnLimitedTier[i] =
+ totalSentRecords > maxBuffersInLimitedTier
+ ? totalSentRecords - maxBuffersInLimitedTier
+ : 0;
+ numExpectedSegmentInUnLimitedTier[i] =
+ numExpectedBuffersInUnLimitedTier[i] / NUM_BUFFERS_IN_A_SEGMENT
+ + (numExpectedBuffersInUnLimitedTier[i] % NUM_BUFFERS_IN_A_SEGMENT == 0
+ ? 0
+ : 1);
+ }
+ }
+
+ private static class TestingTierProducerAgent implements TierProducerAgent {
+ private final boolean isCapacityLimited;
+ private final int numSegmentCapacity;
+ private final int[] numReceivedSegments;
+ private final int[] numTotalReceivedBuffers;
+ private final int[] numReceivedBuffersInCurrentSegment;
+ private boolean isClosed;
+
+ TestingTierProducerAgent(boolean isCapacityLimited, int numSubpartitions) {
+ this(isCapacityLimited, numSubpartitions, Integer.MAX_VALUE);
+ }
+
+ TestingTierProducerAgent(
+ boolean isCapacityLimited, int numSubpartitions, int numSegmentCapacity) {
+ this.isCapacityLimited = isCapacityLimited;
+ this.numSegmentCapacity = numSegmentCapacity;
+ this.numReceivedSegments = new int[numSubpartitions];
+ this.numTotalReceivedBuffers = new int[numSubpartitions];
+ this.numReceivedBuffersInCurrentSegment = new int[numSubpartitions];
+
+ Arrays.fill(numReceivedSegments, 0);
+ Arrays.fill(numTotalReceivedBuffers, 0);
+ Arrays.fill(numReceivedBuffersInCurrentSegment, 0);
+ }
+
+ @Override
+ public boolean tryStartNewSegment(
+ TieredStorageSubpartitionId subpartitionId, int segmentId) {
+ boolean canStartNewSegment =
+ !isCapacityLimited
+ || numReceivedSegments[subpartitionId.getSubpartitionId()]
+ < numSegmentCapacity;
+ if (canStartNewSegment) {
+ numReceivedSegments[subpartitionId.getSubpartitionId()]++;
+ }
+ return canStartNewSegment;
+ }
+
+ @Override
+ public boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer) {
+ if (numReceivedBuffersInCurrentSegment[subpartitionId.getSubpartitionId()] + 1
+ > NUM_BUFFERS_IN_A_SEGMENT) {
+ numReceivedBuffersInCurrentSegment[subpartitionId.getSubpartitionId()] = 0;
+ return false;
+ }
+ numReceivedBuffersInCurrentSegment[subpartitionId.getSubpartitionId()]++;
+ numTotalReceivedBuffers[subpartitionId.getSubpartitionId()]++;
+ finishedBuffer.recycleBuffer();
+ return true;
+ }
+
+ @Override
+ public void close() {
+ this.isClosed = true;
+ }
+ }
+
+ private TieredStorageMemoryManagerImpl createStorageMemoryManager(int numBuffersInBufferPool)
+ throws IOException {
+ BufferPool bufferPool =
+ globalPool.createBufferPool(numBuffersInBufferPool, numBuffersInBufferPool);
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ new TieredStorageMemoryManagerImpl(NUM_BUFFERS_TRIGGER_FLUSH_RATIO, true);
+ storageMemoryManager.setup(
+ bufferPool, Collections.singletonList(new TieredStorageMemorySpec(this, 1)));
+ return storageMemoryManager;
+ }
+
+ private static ByteBuffer generateRandomData(int dataSize, Random random) {
+ byte[] dataWritten = new byte[dataSize];
+ random.nextBytes(dataWritten);
+ return ByteBuffer.wrap(dataWritten);
+ }
Review Comment:
Fixed by introducing a `TieredStorageTestUtils `.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214031417
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java:
##########
@@ -35,9 +41,21 @@ public interface TierProducerAgent {
*/
boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId);
- /** Writes the finished {@link Buffer} to the consumer. */
- boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer)
- throws IOException;
+ /**
+ * Writes the finished {@link Buffer} to the consumer.
+ *
+ * <p>Note that the tier must ensure that the buffer is written successfully without any
+ * exceptions, in order to guarantee that the buffer will be recycled. If this method throws an
+ * exception in the subsequent modifications, the caller should make sure that the buffer is
+ * recycled finally.
Review Comment:
Fixed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java:
##########
@@ -21,9 +21,15 @@
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. */
+/**
+ * The producer-side agent of a Tier.
+ *
+ * <p>Note that when writing a buffer to a tier, the {@link TierProducerAgent} should first call
+ * {@code tryStartNewSegment} to start a new segment. The agent can then continue writing the buffer
+ * to the tier as long as the return value of {@code write} is true. If the return value of {@code
+ * write} is false, it indicates that the current segment can no longer store the buffer, and the
+ * agent should try to start a new segment before writing the buffer.
+ */
public interface TierProducerAgent {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #22652: [FLINK-31640][network] Introduce the segment id tracker for the tiered storage
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1208289916
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTracker.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+
+/**
+ * {@link SubpartitionSegmentIdTracker} is to track segment index for each subpartition. Each {@link
+ * TierProducerAgent}'s data manager has a separate {@link SubpartitionSegmentIdTracker}.
+ */
+public interface SubpartitionSegmentIdTracker {
Review Comment:
I wonder whether it's necessary to separate the interface and implementation in this case.
Usually, we need interface to:
- Provide multiple implementations
- Restrict access to part of the public methods
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTracker.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+
+/**
+ * {@link SubpartitionSegmentIdTracker} is to track segment index for each subpartition. Each {@link
+ * TierProducerAgent}'s data manager has a separate {@link SubpartitionSegmentIdTracker}.
+ */
+public interface SubpartitionSegmentIdTracker {
+
+ /**
+ * Add a new segment index belong to the subpartition.
+ *
+ * @param subpartitionId the subpartition index
+ * @param segmentId the segment index to be added
+ */
+ void addSegmentIndex(TieredStorageSubpartitionId subpartitionId, int segmentId);
+
+ /**
+ * Return whether the {@link SubpartitionSegmentIdTracker} contains the segment index of the
+ * specific subpartition.
+ *
+ * @param subpartitionId the subpartition index
+ * @param segmentId the segment index to be added
+ * @return whether the tracker contains the segment index of the subpartition
+ */
+ boolean hasCurrentSegment(TieredStorageSubpartitionId subpartitionId, int segmentId);
Review Comment:
`hasSegment`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +145,66 @@ private void writeAccumulatedBuffers(
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.
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int tierIndex = chooseStorageTierIfNeeded(subpartitionId);
+
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+ updateStatistics(compressedBuffer);
+ boolean isLastBufferInSegment =
+ tierProducerAgents.get(tierIndex).write(subpartitionId, compressedBuffer);
+ if (isLastBufferInSegment) {
+ isSubpartitionSegmentFinished[subpartitionIndex] = true;
+ }
+ }
+
+ private int chooseStorageTierIfNeeded(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ if (isSubpartitionSegmentFinished[subpartitionIndex]) {
+ // If the current segment is finished, this subpartition need choose a new storage tier.
+ internalChooseStorageTier(subpartitionId);
+ isSubpartitionSegmentFinished[subpartitionIndex] = false;
+ }
+ return currentSubpartitionTierIndex[subpartitionIndex];
+ }
+
+ private void internalChooseStorageTier(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (int tierIndex = 0; tierIndex < tierProducerAgents.size(); ++tierIndex) {
+ if (tierProducerAgents
+ .get(tierIndex)
Review Comment:
This can be simplified if we change `int[] currentSubpartitionTierIndex` to `TierProducerAgent currentSubpartitionTier`. We won't need to loop over the integers then get it from the list.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTracker.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+
+/**
+ * {@link SubpartitionSegmentIdTracker} is to track segment index for each subpartition. Each {@link
+ * TierProducerAgent}'s data manager has a separate {@link SubpartitionSegmentIdTracker}.
+ */
+public interface SubpartitionSegmentIdTracker {
+
+ /**
+ * Add a new segment index belong to the subpartition.
+ *
+ * @param subpartitionId the subpartition index
+ * @param segmentId the segment index to be added
+ */
+ void addSegmentIndex(TieredStorageSubpartitionId subpartitionId, int segmentId);
Review Comment:
`addSegment`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTracker.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+
+/**
+ * {@link SubpartitionSegmentIdTracker} is to track segment index for each subpartition. Each {@link
+ * TierProducerAgent}'s data manager has a separate {@link SubpartitionSegmentIdTracker}.
Review Comment:
What is "data manager"?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -42,6 +45,17 @@ public class TieredStorageProducerClient {
private final List<TierProducerAgent> tierProducerAgents;
+ /** The current writing segment index for each subpartition. */
+ private final int[] currentSubpartitionSegmentId;
+
+ /** The current writing tier index for each subpartition. */
+ private final int[] currentSubpartitionTierIndex;
+
+ /** Indicate whether the segment for each subpartition is finished. */
+ private final boolean[] isSubpartitionSegmentFinished;
Review Comment:
`isCurrentSubpartitionSegmentFinished`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +145,66 @@ private void writeAccumulatedBuffers(
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.
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int tierIndex = chooseStorageTierIfNeeded(subpartitionId);
+
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+ updateStatistics(compressedBuffer);
+ boolean isLastBufferInSegment =
+ tierProducerAgents.get(tierIndex).write(subpartitionId, compressedBuffer);
Review Comment:
`TierProducerAgent#write` lacks docs for the return value semantics. I thought the return-value mean whether the buffer is successfully written. How does the agent know whether it's the last buffer without knowing the size of next buffer?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -80,13 +101,18 @@ public void write(
if (isBroadcast && !isBroadcastOnly) {
for (int i = 0; i < numSubpartitions; ++i) {
- bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ bufferAccumulator.receive(
+ record.duplicate(), new TieredStorageSubpartitionId(i), dataType);
Review Comment:
Let's fix this in a separate hotfix commit.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTrackerImpl.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+import org.apache.flink.util.function.SupplierWithException;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * The implementation of {@link SubpartitionSegmentIdTracker}. Each {@link TierProducerAgent} has a
+ * {@link SubpartitionSegmentIdTrackerImpl} to record the segments belong to the producer agent.
+ */
+public class SubpartitionSegmentIdTrackerImpl implements SubpartitionSegmentIdTracker {
+
+ /**
+ * Each subpartition calculates the amount of data written to a tier separately. If the amount
+ * of data exceeds the threshold, the current writing segment will be finished. Because
+ * different subpartitions may have duplicate segment indexes, this field needs to distinguish
+ * between different subpartitions.
+ */
+ private final Map<TieredStorageSubpartitionId, HashSet<Integer>> subpartitionSegmentIds;
+
+ /** Indicate whether this is a broadcast only partition. */
+ private final Boolean isBroadCastOnly;
+
+ /** Record the latest segment index for each subpartition. */
+ private final int[] latestSegmentIds;
+
+ /** The locks for all subpartitions. */
+ private final Lock[] locks;
+
+ public SubpartitionSegmentIdTrackerImpl(int numSubpartitions, Boolean isBroadcastOnly) {
+ this.isBroadCastOnly = isBroadcastOnly;
+ int effectiveNumSubpartitions = isBroadcastOnly ? 1 : numSubpartitions;
+ this.latestSegmentIds = new int[effectiveNumSubpartitions];
+ this.subpartitionSegmentIds = new HashMap<>();
+ this.locks = new Lock[effectiveNumSubpartitions];
+
+ Arrays.fill(latestSegmentIds, -1);
+ for (int i = 0; i < effectiveNumSubpartitions; i++) {
+ locks[i] = new ReentrantLock();
+ subpartitionSegmentIds.put(new TieredStorageSubpartitionId(i), new HashSet<>());
Review Comment:
This is heavy. Try replace `new HashSet<>()` with `ConcurrentHashMap.newKeySet()`. This basically creates a "ConcurrentHashSet" backed by `ConcurrentHashMap`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +145,66 @@ private void writeAccumulatedBuffers(
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.
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int tierIndex = chooseStorageTierIfNeeded(subpartitionId);
+
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+ updateStatistics(compressedBuffer);
+ boolean isLastBufferInSegment =
+ tierProducerAgents.get(tierIndex).write(subpartitionId, compressedBuffer);
+ if (isLastBufferInSegment) {
+ isSubpartitionSegmentFinished[subpartitionIndex] = true;
+ }
+ }
+
+ private int chooseStorageTierIfNeeded(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ if (isSubpartitionSegmentFinished[subpartitionIndex]) {
+ // If the current segment is finished, this subpartition need choose a new storage tier.
+ internalChooseStorageTier(subpartitionId);
+ isSubpartitionSegmentFinished[subpartitionIndex] = false;
+ }
+ return currentSubpartitionTierIndex[subpartitionIndex];
+ }
+
+ private void internalChooseStorageTier(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (int tierIndex = 0; tierIndex < tierProducerAgents.size(); ++tierIndex) {
Review Comment:
We should make it explicit that `tierProducerAgents` stores agents ordered by priority.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211220042
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTracker.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+
+/**
+ * {@link SubpartitionSegmentIdTracker} is to track segment index for each subpartition. Each {@link
+ * TierProducerAgent}'s data manager has a separate {@link SubpartitionSegmentIdTracker}.
+ */
+public interface SubpartitionSegmentIdTracker {
+
+ /**
+ * Add a new segment index belong to the subpartition.
+ *
+ * @param subpartitionId the subpartition index
+ * @param segmentId the segment index to be added
+ */
+ void addSegmentIndex(TieredStorageSubpartitionId subpartitionId, int segmentId);
Review Comment:
Based on the offline discussion, the segment tracker will be introduced in the subsequential PR.
When introducing the segment tracker, this will be fixed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTracker.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+
+/**
+ * {@link SubpartitionSegmentIdTracker} is to track segment index for each subpartition. Each {@link
+ * TierProducerAgent}'s data manager has a separate {@link SubpartitionSegmentIdTracker}.
+ */
+public interface SubpartitionSegmentIdTracker {
+
+ /**
+ * Add a new segment index belong to the subpartition.
+ *
+ * @param subpartitionId the subpartition index
+ * @param segmentId the segment index to be added
+ */
+ void addSegmentIndex(TieredStorageSubpartitionId subpartitionId, int segmentId);
+
+ /**
+ * Return whether the {@link SubpartitionSegmentIdTracker} contains the segment index of the
+ * specific subpartition.
+ *
+ * @param subpartitionId the subpartition index
+ * @param segmentId the segment index to be added
+ * @return whether the tracker contains the segment index of the subpartition
+ */
+ boolean hasCurrentSegment(TieredStorageSubpartitionId subpartitionId, int segmentId);
Review Comment:
Based on the offline discussion, the segment tracker will be introduced in the subsequential PR.
When introducing the segment tracker, this will be fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211224068
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +145,66 @@ private void writeAccumulatedBuffers(
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.
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int tierIndex = chooseStorageTierIfNeeded(subpartitionId);
+
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+ updateStatistics(compressedBuffer);
+ boolean isLastBufferInSegment =
+ tierProducerAgents.get(tierIndex).write(subpartitionId, compressedBuffer);
+ if (isLastBufferInSegment) {
+ isSubpartitionSegmentFinished[subpartitionIndex] = true;
+ }
+ }
+
+ private int chooseStorageTierIfNeeded(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ if (isSubpartitionSegmentFinished[subpartitionIndex]) {
+ // If the current segment is finished, this subpartition need choose a new storage tier.
+ internalChooseStorageTier(subpartitionId);
+ isSubpartitionSegmentFinished[subpartitionIndex] = false;
+ }
+ return currentSubpartitionTierIndex[subpartitionIndex];
+ }
+
+ private void internalChooseStorageTier(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (int tierIndex = 0; tierIndex < tierProducerAgents.size(); ++tierIndex) {
Review Comment:
Fixed. Added a doc for the tierProducerAgents to show that they are sorted by priority.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212694450
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,11 +128,13 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
+ Queue<Buffer> buffers = new ArrayDeque<>(accumulatedBuffers);
try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ while (!buffers.isEmpty()) {
+ writeAccumulatedBuffer(subpartitionId, buffers.poll());
}
} catch (IOException e) {
+ buffers.forEach(Buffer::recycleBuffer);
ExceptionUtils.rethrow(e);
}
Review Comment:
1. Replaced it with `iterator`.
2. The exception will be thrown only when choosing a tier. So we should recycle the writing buffer in the `writeAccumulatedBuffers`.
In `writeAccumulatedBuffer`, the `write` will not throw an exception currently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214224183
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingTierProducerAgent;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.TieredStorageTestUtils.generateRandomData;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(numSubpartitions, Collections.emptyList());
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testWriteRecords() throws IOException {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ int maxNumToWriteRecordsPerSubpartition = 1000;
+ Random random = new Random();
+
+ TestingTierProducerAgent tierProducerAgent = new TestingTierProducerAgent();
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> true);
+ tierProducerAgent.setTryWriteReturnValueSupplier(() -> new Boolean[] {false, true});
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(
+ numSubpartitions, Collections.singletonList(tierProducerAgent));
+
+ int numWriteRecords = 0;
+ for (int j = 0; j < numSubpartitions; j++) {
+ for (int i = 0; i < random.nextInt(maxNumToWriteRecordsPerSubpartition); i++) {
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(j),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast);
+ numWriteRecords++;
+ }
+ }
+
+ int numExpectedBuffers = isBroadcast ? numWriteRecords * numSubpartitions : numWriteRecords;
+ assertThat(tierProducerAgent.numTotalReceivedBuffers()).isEqualTo(numExpectedBuffers);
+ }
+
+ @TestTemplate
+ void testTierCanNotStartNewSegment() throws IOException {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TestingTierProducerAgent tierProducerAgent = new TestingTierProducerAgent();
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> false);
+ tierProducerAgent.setTryWriteReturnValueSupplier(() -> new Boolean[] {false, true});
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(
+ numSubpartitions, Collections.singletonList(tierProducerAgent));
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> true);
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast);
+
+ assertThat(tierProducerAgent.numTotalReceivedBuffers())
+ .isEqualTo(isBroadcast ? numSubpartitions : 1);
+ }
+
+ @TestTemplate
+ void testClose() {
+ int numSubpartitions = 10;
+
+ TestingTierProducerAgent tierProducerAgent = new TestingTierProducerAgent();
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> true);
+ tierProducerAgent.setTryWriteReturnValueSupplier(() -> new Boolean[] {false, true});
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on PR #22652:
URL: https://github.com/apache/flink/pull/22652#issuecomment-1571493540
@xintongsong Thanks for the careful review. I have addressed the comments with two fix-up commits.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212863225
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -80,13 +107,23 @@ public void write(
if (isBroadcast && !isBroadcastOnly) {
for (int i = 0; i < numSubpartitions; ++i) {
- bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ // As the tiered storage subpartition ID is created only for broadcast records,
+ // which are fewer than normal records, the performance impact of generating new
+ // TieredStorageSubpartitionId objects is expected to be manageable. If the
+ // performance is significantly affected, this logic will be optimized accordingly.
+ bufferAccumulator.receive(
+ record.duplicate(), new TieredStorageSubpartitionId(i), dataType);
}
} else {
bufferAccumulator.receive(record, subpartitionId, dataType);
}
}
+ public void setMetricStatisticsUpdater(
+ Consumer<TieredStorageProducerMetricUpdate> metricStatisticsUpdater) {
+ this.metricStatisticsUpdater = metricStatisticsUpdater;
Review Comment:
```suggestion
this.metricStatisticsUpdater = checkNotNull(metricStatisticsUpdater);
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java:
##########
@@ -21,9 +21,15 @@
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. */
+/**
+ * The producer-side agent of a Tier.
+ *
+ * <p>Note that when writing a buffer to a tier, the {@link TierProducerAgent} should first call
+ * {@code tryStartNewSegment} to start a new segment. The agent can then continue writing the buffer
+ * to the tier as long as the return value of {@code write} is true. If the return value of {@code
+ * write} is false, it indicates that the current segment can no longer store the buffer, and the
+ * agent should try to start a new segment before writing the buffer.
+ */
public interface TierProducerAgent {
Review Comment:
I'd prefer let this extends `AutoClosable`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java:
##########
@@ -35,9 +41,21 @@ public interface TierProducerAgent {
*/
boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId);
- /** Writes the finished {@link Buffer} to the consumer. */
- boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer)
- throws IOException;
+ /**
+ * Writes the finished {@link Buffer} to the consumer.
+ *
+ * <p>Note that the tier must ensure that the buffer is written successfully without any
+ * exceptions, in order to guarantee that the buffer will be recycled. If this method throws an
+ * exception in the subsequent modifications, the caller should make sure that the buffer is
+ * recycled finally.
+ *
+ * @param subpartitionId the subpartition id that the buffer is writing to
+ * @param finishedBuffer the writing buffer
+ * @return return true if the buffer is written successfully, return false if the current
+ * segment can not store this buffer and the current segment is finished. When returning
+ * false, the agent should try start a new segment before writing the buffer.
+ */
+ boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer);
Review Comment:
how about rename this to `tryWrite`?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, Collections.emptyList());
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testEmptyMetricUpdater() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+ TestingTierProducerAgent testingTierProducerAgent =
+ new TestingTierProducerAgent(false, numSubpartitions);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions,
+ false,
+ bufferAccumulator,
+ null,
+ Collections.singletonList(testingTierProducerAgent));
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ @TestTemplate
+ void testWriteRecordsToMultipleStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int numSegmentCapacityInLimitedTier = 12;
+ int numMaxToSendRecords = 200;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+ List<TierProducerAgent> tierProducerAgents = new ArrayList<>();
+ TestingTierProducerAgent limitedSegmentsTierProducerAgent =
+ new TestingTierProducerAgent(
+ true, numSubpartitions, numSegmentCapacityInLimitedTier);
+ TestingTierProducerAgent unLimitedTierProducerAgent =
+ new TestingTierProducerAgent(false, numSubpartitions);
+ tierProducerAgents.add(limitedSegmentsTierProducerAgent);
+ tierProducerAgents.add(unLimitedTierProducerAgent);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, tierProducerAgents);
+
+ AtomicInteger numWriteBuffers = new AtomicInteger();
+ AtomicInteger numWriteBytes = new AtomicInteger();
+ tieredStorageProducerClient.setMetricStatisticsUpdater(
+ metricStatistics -> {
+ numWriteBuffers.set(
+ numWriteBuffers.get() + metricStatistics.numWriteBuffersDelta());
+ numWriteBytes.set(numWriteBytes.get() + metricStatistics.numWriteBytesDelta());
+ });
+
+ int[] numSentRecords = new int[numSubpartitions];
+ for (int i = 0; i < numSubpartitions; i++) {
+ int numToSend = random.nextInt(numMaxToSendRecords - 1) + 1;
+ numSentRecords[i] = numToSend;
+ TieredStorageSubpartitionId subpartitionId = new TieredStorageSubpartitionId(i);
+ for (int j = 0; j < numToSend; j++) {
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ subpartitionId,
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast);
+ }
+ }
+
+ checkReceivedBuffersAndSegments(
+ numSubpartitions,
+ bufferSize,
+ numSegmentCapacityInLimitedTier,
+ limitedSegmentsTierProducerAgent,
+ unLimitedTierProducerAgent,
+ numSentRecords,
+ numWriteBuffers,
+ numWriteBytes);
+
+ assertThat(limitedSegmentsTierProducerAgent.isClosed).isFalse();
+ assertThat(unLimitedTierProducerAgent.isClosed).isFalse();
+ tieredStorageProducerClient.close();
+ assertThat(limitedSegmentsTierProducerAgent.isClosed).isTrue();
+ assertThat(unLimitedTierProducerAgent.isClosed).isTrue();
+ storageMemoryManager.release();
+ }
+
+ private void checkReceivedBuffersAndSegments(
+ int numSubpartitions,
+ int bufferSize,
+ int numSegmentCapacityInLimitedTier,
+ TestingTierProducerAgent limitedSegmentsTierProducerAgent,
+ TestingTierProducerAgent unLimitedTierProducerAgent,
+ int[] numSentRecords,
+ AtomicInteger numWriteBuffers,
+ AtomicInteger numWriteBytes) {
+ int[] numExpectedBuffersInLimitedTier = new int[numSubpartitions];
+ int[] numExpectedBuffersInUnLimitedTier = new int[numSubpartitions];
+ int[] numExpectedSegmentInLimitedTier = new int[numSubpartitions];
+ int[] numExpectedSegmentInUnLimitedTier = new int[numSubpartitions];
+
+ calculateNumExpectedBuffersAndSegments(
+ numSubpartitions,
+ numSegmentCapacityInLimitedTier,
+ numSentRecords,
+ numExpectedBuffersInLimitedTier,
+ numExpectedBuffersInUnLimitedTier,
+ numExpectedSegmentInLimitedTier,
+ numExpectedSegmentInUnLimitedTier);
+
+ for (int i = 0; i < numSubpartitions; i++) {
+ assertThat(limitedSegmentsTierProducerAgent.numTotalReceivedBuffers[i])
+ .isEqualTo(numExpectedBuffersInLimitedTier[i]);
+ assertThat(unLimitedTierProducerAgent.numTotalReceivedBuffers[i])
+ .isEqualTo(numExpectedBuffersInUnLimitedTier[i]);
+ assertThat(limitedSegmentsTierProducerAgent.numTotalReceivedBuffers[i] > 0).isTrue();
+ assertThat(limitedSegmentsTierProducerAgent.numReceivedSegments[i])
+ .isEqualTo(numExpectedSegmentInLimitedTier[i]);
+ assertThat(unLimitedTierProducerAgent.numReceivedSegments[i])
+ .isEqualTo(numExpectedSegmentInUnLimitedTier[i]);
+ }
+
+ int numTotalRecords = Arrays.stream(numSentRecords).sum();
+ int numExpectedBuffers = isBroadcast ? numTotalRecords * numSubpartitions : numTotalRecords;
+ assertThat(numWriteBuffers.get()).isEqualTo(numExpectedBuffers);
+ assertThat(numWriteBytes.get()).isEqualTo(numExpectedBuffers * bufferSize);
+ }
+
+ private void calculateNumExpectedBuffersAndSegments(
+ int numSubpartitions,
+ int numSegmentCapacityInLimitedTier,
+ int[] numSentRecords,
+ int[] numExpectedBuffersInLimitedTier,
+ int[] numExpectedBuffersInUnLimitedTier,
+ int[] numExpectedSegmentInLimitedTier,
+ int[] numExpectedSegmentInUnLimitedTier) {
+ int numTotalRecords = Arrays.stream(numSentRecords).map(Integer::new).sum();
+ for (int i = 0; i < numSubpartitions; i++) {
+ int totalSentRecords = isBroadcast ? numTotalRecords : numSentRecords[i];
+ int maxBuffersInLimitedTier =
+ numSegmentCapacityInLimitedTier * NUM_BUFFERS_IN_A_SEGMENT;
+
+ numExpectedBuffersInLimitedTier[i] =
+ Math.min(totalSentRecords, maxBuffersInLimitedTier);
+ numExpectedSegmentInLimitedTier[i] =
+ numExpectedBuffersInLimitedTier[i] / NUM_BUFFERS_IN_A_SEGMENT
+ + (numExpectedBuffersInLimitedTier[i] % NUM_BUFFERS_IN_A_SEGMENT == 0
+ ? 0
+ : 1);
+ numExpectedBuffersInUnLimitedTier[i] =
+ totalSentRecords > maxBuffersInLimitedTier
+ ? totalSentRecords - maxBuffersInLimitedTier
+ : 0;
+ numExpectedSegmentInUnLimitedTier[i] =
+ numExpectedBuffersInUnLimitedTier[i] / NUM_BUFFERS_IN_A_SEGMENT
+ + (numExpectedBuffersInUnLimitedTier[i] % NUM_BUFFERS_IN_A_SEGMENT == 0
+ ? 0
+ : 1);
+ }
+ }
+
+ private static class TestingTierProducerAgent implements TierProducerAgent {
+ private final boolean isCapacityLimited;
+ private final int numSegmentCapacity;
+ private final int[] numReceivedSegments;
+ private final int[] numTotalReceivedBuffers;
+ private final int[] numReceivedBuffersInCurrentSegment;
+ private boolean isClosed;
+
+ TestingTierProducerAgent(boolean isCapacityLimited, int numSubpartitions) {
+ this(isCapacityLimited, numSubpartitions, Integer.MAX_VALUE);
+ }
+
+ TestingTierProducerAgent(
+ boolean isCapacityLimited, int numSubpartitions, int numSegmentCapacity) {
+ this.isCapacityLimited = isCapacityLimited;
+ this.numSegmentCapacity = numSegmentCapacity;
+ this.numReceivedSegments = new int[numSubpartitions];
+ this.numTotalReceivedBuffers = new int[numSubpartitions];
+ this.numReceivedBuffersInCurrentSegment = new int[numSubpartitions];
+
+ Arrays.fill(numReceivedSegments, 0);
+ Arrays.fill(numTotalReceivedBuffers, 0);
+ Arrays.fill(numReceivedBuffersInCurrentSegment, 0);
+ }
+
+ @Override
+ public boolean tryStartNewSegment(
+ TieredStorageSubpartitionId subpartitionId, int segmentId) {
+ boolean canStartNewSegment =
+ !isCapacityLimited
+ || numReceivedSegments[subpartitionId.getSubpartitionId()]
+ < numSegmentCapacity;
+ if (canStartNewSegment) {
+ numReceivedSegments[subpartitionId.getSubpartitionId()]++;
+ }
+ return canStartNewSegment;
+ }
+
+ @Override
+ public boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer) {
+ if (numReceivedBuffersInCurrentSegment[subpartitionId.getSubpartitionId()] + 1
+ > NUM_BUFFERS_IN_A_SEGMENT) {
+ numReceivedBuffersInCurrentSegment[subpartitionId.getSubpartitionId()] = 0;
+ return false;
+ }
+ numReceivedBuffersInCurrentSegment[subpartitionId.getSubpartitionId()]++;
+ numTotalReceivedBuffers[subpartitionId.getSubpartitionId()]++;
+ finishedBuffer.recycleBuffer();
+ return true;
+ }
+
+ @Override
+ public void close() {
+ this.isClosed = true;
+ }
+ }
+
+ private TieredStorageMemoryManagerImpl createStorageMemoryManager(int numBuffersInBufferPool)
+ throws IOException {
+ BufferPool bufferPool =
+ globalPool.createBufferPool(numBuffersInBufferPool, numBuffersInBufferPool);
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ new TieredStorageMemoryManagerImpl(NUM_BUFFERS_TRIGGER_FLUSH_RATIO, true);
+ storageMemoryManager.setup(
+ bufferPool, Collections.singletonList(new TieredStorageMemorySpec(this, 1)));
+ return storageMemoryManager;
+ }
+
+ private static ByteBuffer generateRandomData(int dataSize, Random random) {
+ byte[] dataWritten = new byte[dataSize];
+ random.nextBytes(dataWritten);
+ return ByteBuffer.wrap(dataWritten);
+ }
Review Comment:
These two methods seem exactly the same as those in `HashBufferAccumulatorTest`, so why not extract them all into a common utils class called `TieredStorageTestUtils`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,26 +137,95 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
- try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ Iterator<Buffer> bufferIterator = accumulatedBuffers.iterator();
+
+ int numWriteBytes = 0;
+ int numWriteBuffers = 0;
+ while (bufferIterator.hasNext()) {
+ Buffer buffer = bufferIterator.next();
+ try {
+ writeAccumulatedBuffer(subpartitionId, buffer);
+ } catch (IOException ioe) {
+ buffer.recycleBuffer();
+ while (bufferIterator.hasNext()) {
+ bufferIterator.next().recycleBuffer();
+ }
+ ExceptionUtils.rethrow(ioe);
}
- } catch (IOException e) {
- ExceptionUtils.rethrow(e);
+ numWriteBuffers++;
+ numWriteBytes += buffer.readableBytes();
}
+ updateMetricStatistics(numWriteBuffers, numWriteBytes);
}
/**
* 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.
*
+ * <p>Note that the method only throws an exception when choosing a storage tier, so the caller
+ * should ensure that the buffer is recycled when throwing an exception.
+ *
* @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.
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ if (!isSuccess) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ checkState(isSuccess, "Failed to write the first buffer to the new segment");
+ }
Review Comment:
```suggestion
if (!currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
subpartitionId, compressedBuffer)) {
chooseStorageTierToStartSegment(subpartitionId);
checkState(
currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
subpartitionId, compressedBuffer),
"Failed to write the first buffer to the new segment");
}
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, Collections.emptyList());
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testEmptyMetricUpdater() throws IOException {
Review Comment:
TBH, I don't think this is a meaningful test case. If we don't set the value for a not initialized field, it will be `null`, which doesn't seem to require special testing.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
Review Comment:
```suggestion
private static final int NUM_BUFFERS_PER_SEGMENT = 5;
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/TierProducerAgent.java:
##########
@@ -35,9 +41,21 @@ public interface TierProducerAgent {
*/
boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId);
- /** Writes the finished {@link Buffer} to the consumer. */
- boolean write(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer)
- throws IOException;
+ /**
+ * Writes the finished {@link Buffer} to the consumer.
+ *
+ * <p>Note that the tier must ensure that the buffer is written successfully without any
+ * exceptions, in order to guarantee that the buffer will be recycled. If this method throws an
+ * exception in the subsequent modifications, the caller should make sure that the buffer is
+ * recycled finally.
Review Comment:
> If this method throws an exception in the subsequent modifications, the caller should make sure that the buffer is recycled finally.
The readability of this paragraph is not good, which has caused me some confusion. UUIC, what we want to express is that if the method is successfully executed (without throwing any exception), the buffer should be released by the caller, otherwise it is the responsibility of the tier.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, Collections.emptyList());
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
Review Comment:
It is difficult to see the connection between this call(i.e. `tieredStorageProducerClient.write`) and this expected exception, I have to debug the code to understand the logic. And this test relies too heavily on the implementation of `BufferAccumulator`, if we change its code in the future, it is likely to implicate this test.
I'd suggestion using a mock `TestingBufferAccumulator` to ensure that this call will flush a finished buffer.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
Review Comment:
These two fields can be promoted to class members and initialized in `before()`.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, Collections.emptyList());
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testEmptyMetricUpdater() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+ TestingTierProducerAgent testingTierProducerAgent =
+ new TestingTierProducerAgent(false, numSubpartitions);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions,
+ false,
+ bufferAccumulator,
+ null,
+ Collections.singletonList(testingTierProducerAgent));
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ @TestTemplate
+ void testWriteRecordsToMultipleStorageTiers() throws IOException {
Review Comment:
This test is too complex, and the some test logic does not belong to `TieredStorageProducerClient`. What we need to test is the logic that only belongs `TieredStorageProducerClient`. Others like whether the buffer is written to the tier is not its concern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211222672
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTracker.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+
+/**
+ * {@link SubpartitionSegmentIdTracker} is to track segment index for each subpartition. Each {@link
+ * TierProducerAgent}'s data manager has a separate {@link SubpartitionSegmentIdTracker}.
Review Comment:
Based on the offline discussion, the segment tracker will be introduced in the subsequential PR.
When introducing the segment tracker, this will be fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214034785
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214045783
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -80,13 +107,23 @@ public void write(
if (isBroadcast && !isBroadcastOnly) {
for (int i = 0; i < numSubpartitions; ++i) {
- bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ // As the tiered storage subpartition ID is created only for broadcast records,
+ // which are fewer than normal records, the performance impact of generating new
+ // TieredStorageSubpartitionId objects is expected to be manageable. If the
+ // performance is significantly affected, this logic will be optimized accordingly.
+ bufferAccumulator.receive(
+ record.duplicate(), new TieredStorageSubpartitionId(i), dataType);
}
} else {
bufferAccumulator.receive(record, subpartitionId, dataType);
}
}
+ public void setMetricStatisticsUpdater(
+ Consumer<TieredStorageProducerMetricUpdate> metricStatisticsUpdater) {
+ this.metricStatisticsUpdater = metricStatisticsUpdater;
Review Comment:
It seems that this line is not updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa merged pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa merged PR #22652:
URL: https://github.com/apache/flink/pull/22652
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212796022
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,26 +137,95 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
- try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ Iterator<Buffer> bufferIterator = accumulatedBuffers.iterator();
+
+ int numWriteBytes = 0;
+ int numWriteBuffers = 0;
+ while (bufferIterator.hasNext()) {
+ Buffer buffer = bufferIterator.next();
+ try {
+ writeAccumulatedBuffer(subpartitionId, buffer);
+ } catch (IOException ioe) {
+ buffer.recycleBuffer();
+ while (bufferIterator.hasNext()) {
+ bufferIterator.next().recycleBuffer();
+ }
+ ExceptionUtils.rethrow(ioe);
}
- } catch (IOException e) {
- ExceptionUtils.rethrow(e);
+ numWriteBuffers++;
+ numWriteBytes += buffer.readableBytes();
}
+ updateMetricStatistics(numWriteBuffers, numWriteBytes);
}
/**
* 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.
*
+ * <p>Note that the method only throws an exception when choosing a storage tier, so the caller
+ * should ensure that the buffer is recycled when throwing an exception.
+ *
* @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.
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ if (!isSuccess) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ checkState(isSuccess, "Failed to write the first buffer to the new segment");
+ }
+ }
+
+ private void chooseStorageTierToStartSegment(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (TierProducerAgent tierProducerAgent : tierProducerAgents) {
+ if (tierProducerAgent.tryStartNewSegment(subpartitionId, nextSegmentIndex)) {
+ // Update the segment index and the chosen storage tier for the subpartition.
+ currentSubpartitionSegmentId[subpartitionIndex] = nextSegmentIndex;
+ currentSubpartitionTierAgent[subpartitionIndex] = tierProducerAgent;
+ return;
+ }
+ }
+ throw new IOException("Failed to choose a storage tier to start a new segment.");
+ }
+
+ private Buffer compressBufferIfPossible(Buffer buffer) {
+ if (!canBeCompressed(buffer)) {
+ return buffer;
+ }
+
+ return checkNotNull(bufferCompressor).compressToOriginalBuffer(buffer);
+ }
+
+ /**
+ * Whether the buffer can be compressed or not. Note that event is not compressed because it is
+ * usually small and the size can become even larger after compression.
+ */
+ private boolean canBeCompressed(Buffer buffer) {
+ return bufferCompressor != null && buffer.isBuffer() && buffer.readableBytes() > 0;
+ }
+
+ private void updateMetricStatistics(int numWriteBuffers, int numWriteBytes) {
+ if (metricStatisticsUpdater != null) {
Review Comment:
Fixed.
In addition, added a test for checking the empty metric updater.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211223311
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -42,6 +45,17 @@ public class TieredStorageProducerClient {
private final List<TierProducerAgent> tierProducerAgents;
+ /** The current writing segment index for each subpartition. */
+ private final int[] currentSubpartitionSegmentId;
+
+ /** The current writing tier index for each subpartition. */
+ private final int[] currentSubpartitionTierIndex;
+
+ /** Indicate whether the segment for each subpartition is finished. */
+ private final boolean[] isSubpartitionSegmentFinished;
Review Comment:
This field is removed because the logic of EndOfSegment has been changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211223739
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -80,13 +101,18 @@ public void write(
if (isBroadcast && !isBroadcastOnly) {
for (int i = 0; i < numSubpartitions; ++i) {
- bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ bufferAccumulator.receive(
+ record.duplicate(), new TieredStorageSubpartitionId(i), dataType);
Review Comment:
OK, Fixed it with a separate commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212751146
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -80,13 +107,23 @@ public void write(
if (isBroadcast && !isBroadcastOnly) {
for (int i = 0; i < numSubpartitions; ++i) {
- bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ // As the tiered storage subpartition ID is created only for broadcast records,
+ // which are fewer than normal records, the performance impact is expected to be
Review Comment:
"the performance impact of generating new `TieredStorageSubpartitionId` objects"
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,26 +137,95 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
- try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ Iterator<Buffer> bufferIterator = accumulatedBuffers.iterator();
+
+ int numWriteBytes = 0;
+ int numWriteBuffers = 0;
+ while (bufferIterator.hasNext()) {
+ Buffer buffer = bufferIterator.next();
+ try {
+ writeAccumulatedBuffer(subpartitionId, buffer);
+ } catch (IOException ioe) {
+ buffer.recycleBuffer();
+ while (bufferIterator.hasNext()) {
+ bufferIterator.next().recycleBuffer();
+ }
+ ExceptionUtils.rethrow(ioe);
}
- } catch (IOException e) {
- ExceptionUtils.rethrow(e);
+ numWriteBuffers++;
+ numWriteBytes += buffer.readableBytes();
}
+ updateMetricStatistics(numWriteBuffers, numWriteBytes);
}
/**
* 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.
*
+ * <p>Note that the method only throws an exception when choosing a storage tier, so the caller
+ * should ensure that the buffer is recycled when throwing an exception.
+ *
* @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.
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ if (!isSuccess) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ checkState(isSuccess, "Failed to write the first buffer to the new segment");
+ }
+ }
+
+ private void chooseStorageTierToStartSegment(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (TierProducerAgent tierProducerAgent : tierProducerAgents) {
+ if (tierProducerAgent.tryStartNewSegment(subpartitionId, nextSegmentIndex)) {
+ // Update the segment index and the chosen storage tier for the subpartition.
+ currentSubpartitionSegmentId[subpartitionIndex] = nextSegmentIndex;
+ currentSubpartitionTierAgent[subpartitionIndex] = tierProducerAgent;
+ return;
+ }
+ }
+ throw new IOException("Failed to choose a storage tier to start a new segment.");
+ }
+
+ private Buffer compressBufferIfPossible(Buffer buffer) {
+ if (!canBeCompressed(buffer)) {
+ return buffer;
+ }
+
+ return checkNotNull(bufferCompressor).compressToOriginalBuffer(buffer);
+ }
+
+ /**
+ * Whether the buffer can be compressed or not. Note that event is not compressed because it is
+ * usually small and the size can become even larger after compression.
+ */
+ private boolean canBeCompressed(Buffer buffer) {
+ return bufferCompressor != null && buffer.isBuffer() && buffer.readableBytes() > 0;
+ }
+
+ private void updateMetricStatistics(int numWriteBuffers, int numWriteBytes) {
Review Comment:
```suggestion
private void updateMetricStatistics(int numWriteBuffersDelta, int numWriteBytesDelta) {
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,26 +137,95 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
- try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ Iterator<Buffer> bufferIterator = accumulatedBuffers.iterator();
+
+ int numWriteBytes = 0;
+ int numWriteBuffers = 0;
+ while (bufferIterator.hasNext()) {
+ Buffer buffer = bufferIterator.next();
+ try {
+ writeAccumulatedBuffer(subpartitionId, buffer);
+ } catch (IOException ioe) {
+ buffer.recycleBuffer();
+ while (bufferIterator.hasNext()) {
+ bufferIterator.next().recycleBuffer();
+ }
+ ExceptionUtils.rethrow(ioe);
}
- } catch (IOException e) {
- ExceptionUtils.rethrow(e);
+ numWriteBuffers++;
+ numWriteBytes += buffer.readableBytes();
}
+ updateMetricStatistics(numWriteBuffers, numWriteBytes);
}
/**
* 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.
*
+ * <p>Note that the method only throws an exception when choosing a storage tier, so the caller
+ * should ensure that the buffer is recycled when throwing an exception.
+ *
* @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.
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ if (!isSuccess) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ checkState(isSuccess, "Failed to write the first buffer to the new segment");
+ }
+ }
+
+ private void chooseStorageTierToStartSegment(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (TierProducerAgent tierProducerAgent : tierProducerAgents) {
+ if (tierProducerAgent.tryStartNewSegment(subpartitionId, nextSegmentIndex)) {
+ // Update the segment index and the chosen storage tier for the subpartition.
+ currentSubpartitionSegmentId[subpartitionIndex] = nextSegmentIndex;
+ currentSubpartitionTierAgent[subpartitionIndex] = tierProducerAgent;
+ return;
+ }
+ }
+ throw new IOException("Failed to choose a storage tier to start a new segment.");
+ }
+
+ private Buffer compressBufferIfPossible(Buffer buffer) {
+ if (!canBeCompressed(buffer)) {
+ return buffer;
+ }
+
+ return checkNotNull(bufferCompressor).compressToOriginalBuffer(buffer);
+ }
+
+ /**
+ * Whether the buffer can be compressed or not. Note that event is not compressed because it is
+ * usually small and the size can become even larger after compression.
+ */
+ private boolean canBeCompressed(Buffer buffer) {
+ return bufferCompressor != null && buffer.isBuffer() && buffer.readableBytes() > 0;
+ }
+
+ private void updateMetricStatistics(int numWriteBuffers, int numWriteBytes) {
+ if (metricStatisticsUpdater != null) {
Review Comment:
I think we should require the updater to be set before writing records. Otherwise, the statistics would be incorrect, which is also hard to discover.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerMetricStatistics.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.storage;
+
+/** The metric statistics for the tiered storage producer. */
+public class TieredStorageProducerMetricStatistics {
Review Comment:
```suggestion
public class TieredStorageProducerStatisticUpdate {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on pull request #22652: [FLINK-31640][network] Introduce the segment id tracker for the tiered storage
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on PR #22652:
URL: https://github.com/apache/flink/pull/22652#issuecomment-1562462769
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211219669
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/SubpartitionSegmentIdTracker.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.TieredStorageSubpartitionId;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierProducerAgent;
+
+/**
+ * {@link SubpartitionSegmentIdTracker} is to track segment index for each subpartition. Each {@link
+ * TierProducerAgent}'s data manager has a separate {@link SubpartitionSegmentIdTracker}.
+ */
+public interface SubpartitionSegmentIdTracker {
Review Comment:
Based on the offline discussion, the segment tracker will be introduced in the subsequential PR.
When introducing the segment tracker, this will be fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212515884
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +149,60 @@ private void writeAccumulatedBuffers(
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.
+ updateStatistics(accumulatedBuffer);
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
Review Comment:
The semantic of `TierProducerAgent#write` return value needs to be documented. E.g., after a write failure, do we allow writing to the segment again?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,11 +128,13 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
+ Queue<Buffer> buffers = new ArrayDeque<>(accumulatedBuffers);
try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ while (!buffers.isEmpty()) {
+ writeAccumulatedBuffer(subpartitionId, buffers.poll());
}
} catch (IOException e) {
+ buffers.forEach(Buffer::recycleBuffer);
ExceptionUtils.rethrow(e);
}
Review Comment:
There's no need to create the `ArrayDeque`. This can be achieved with `List#iterator()`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,11 +128,13 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
+ Queue<Buffer> buffers = new ArrayDeque<>(accumulatedBuffers);
try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ while (!buffers.isEmpty()) {
+ writeAccumulatedBuffer(subpartitionId, buffers.poll());
}
} catch (IOException e) {
+ buffers.forEach(Buffer::recycleBuffer);
ExceptionUtils.rethrow(e);
}
Review Comment:
What may cause an exception here? What happens to the particular buffer that causes the exception?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +149,60 @@ private void writeAccumulatedBuffers(
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.
+ updateStatistics(accumulatedBuffer);
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ if (!isSuccess) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ checkState(isSuccess, "Failed to write the first buffer to the new segment");
+ }
+ }
+
+ private void chooseStorageTierToStartSegment(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (TierProducerAgent tierProducerAgent : tierProducerAgents) {
+ if (tierProducerAgent.tryStartNewSegment(subpartitionId, nextSegmentIndex)) {
+ // Update the segment index and the chosen storage tier for the subpartition.
+ currentSubpartitionSegmentId[subpartitionIndex] = nextSegmentIndex;
+ currentSubpartitionTierAgent[subpartitionIndex] = tierProducerAgent;
+ return;
+ }
+ }
+ throw new IOException("Failed to choose a storage tier to start a new segment.");
+ }
+
+ private Buffer compressBufferIfPossible(Buffer buffer) {
+ if (!canBeCompressed(buffer)) {
+ return buffer;
+ }
+
+ return checkNotNull(bufferCompressor).compressToOriginalBuffer(buffer);
+ }
+
+ /**
+ * Whether the buffer can be compressed or not. Note that event is not compressed because it is
+ * usually small and the size can become even larger after compression.
+ */
+ private boolean canBeCompressed(Buffer buffer) {
+ return bufferCompressor != null && buffer.isBuffer() && buffer.readableBytes() > 0;
+ }
+
+ private void updateStatistics(Buffer buffer) {
+ checkNotNull(outputMetrics).getNumBuffersOut().inc();
+ checkNotNull(outputMetrics).getNumBytesOut().inc(buffer.readableBytes());
Review Comment:
This indicates Flink's metric mechanism invading tiered storage. Alternatively, we can apply a listener pattern that tiered result partition subscribes to the tiered storage statistic updates.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214030893
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -100,26 +137,95 @@ public void close() {
*/
private void writeAccumulatedBuffers(
TieredStorageSubpartitionId subpartitionId, List<Buffer> accumulatedBuffers) {
- try {
- for (Buffer finishedBuffer : accumulatedBuffers) {
- writeAccumulatedBuffer(subpartitionId, finishedBuffer);
+ Iterator<Buffer> bufferIterator = accumulatedBuffers.iterator();
+
+ int numWriteBytes = 0;
+ int numWriteBuffers = 0;
+ while (bufferIterator.hasNext()) {
+ Buffer buffer = bufferIterator.next();
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on PR #22652:
URL: https://github.com/apache/flink/pull/22652#issuecomment-1573304176
@reswqa Thanks for helping review, I have addressed the comments. Could you help take a look again?
Due to the significant code changes in the test code, I didn't use a fix-up mode to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on PR #22652:
URL: https://github.com/apache/flink/pull/22652#issuecomment-1573354370
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212697793
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +149,60 @@ private void writeAccumulatedBuffers(
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.
+ updateStatistics(accumulatedBuffer);
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ if (!isSuccess) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
+ subpartitionId, compressedBuffer);
+ checkState(isSuccess, "Failed to write the first buffer to the new segment");
+ }
+ }
+
+ private void chooseStorageTierToStartSegment(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (TierProducerAgent tierProducerAgent : tierProducerAgents) {
+ if (tierProducerAgent.tryStartNewSegment(subpartitionId, nextSegmentIndex)) {
+ // Update the segment index and the chosen storage tier for the subpartition.
+ currentSubpartitionSegmentId[subpartitionIndex] = nextSegmentIndex;
+ currentSubpartitionTierAgent[subpartitionIndex] = tierProducerAgent;
+ return;
+ }
+ }
+ throw new IOException("Failed to choose a storage tier to start a new segment.");
+ }
+
+ private Buffer compressBufferIfPossible(Buffer buffer) {
+ if (!canBeCompressed(buffer)) {
+ return buffer;
+ }
+
+ return checkNotNull(bufferCompressor).compressToOriginalBuffer(buffer);
+ }
+
+ /**
+ * Whether the buffer can be compressed or not. Note that event is not compressed because it is
+ * usually small and the size can become even larger after compression.
+ */
+ private boolean canBeCompressed(Buffer buffer) {
+ return bufferCompressor != null && buffer.isBuffer() && buffer.readableBytes() > 0;
+ }
+
+ private void updateStatistics(Buffer buffer) {
+ checkNotNull(outputMetrics).getNumBuffersOut().inc();
+ checkNotNull(outputMetrics).getNumBytesOut().inc(buffer.readableBytes());
Review Comment:
Fixed by introducing a `Consumer<TieredStorageProducerMetricStatistics> metricStatisticsUpdater` in the producer client.
The `TieredResultPartition` will listen to the metric updater.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214057512
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -80,13 +107,23 @@ public void write(
if (isBroadcast && !isBroadcastOnly) {
for (int i = 0; i < numSubpartitions; ++i) {
- bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ // As the tiered storage subpartition ID is created only for broadcast records,
+ // which are fewer than normal records, the performance impact of generating new
+ // TieredStorageSubpartitionId objects is expected to be manageable. If the
+ // performance is significantly affected, this logic will be optimized accordingly.
+ bufferAccumulator.receive(
+ record.duplicate(), new TieredStorageSubpartitionId(i), dataType);
}
} else {
bufferAccumulator.receive(record, subpartitionId, dataType);
}
}
+ public void setMetricStatisticsUpdater(
+ Consumer<TieredStorageProducerMetricUpdate> metricStatisticsUpdater) {
+ this.metricStatisticsUpdater = metricStatisticsUpdater;
Review Comment:
Ah. I missed this. Fixed this now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214057806
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
Review Comment:
```suggestion
private Supplier<Boolean> tryStartNewSegmentSupplier;
```
In general, for mock class, we named it's fields as follows:
`MethodName+(Runnable/Supplier/Consumer/Function)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214036360
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, Collections.emptyList());
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
Review Comment:
Fixed. Used the `TestingBufferAccumulator `
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ new TieredStorageProducerClient(
+ numSubpartitions, false, bufferAccumulator, null, Collections.emptyList());
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testEmptyMetricUpdater() throws IOException {
Review Comment:
Removed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214035752
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,352 @@
+/*
+ * 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.BufferPool;
+import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private static final float NUM_BUFFERS_TRIGGER_FLUSH_RATIO = 0.6f;
+
+ private static final int NUM_BUFFERS_IN_A_SEGMENT = 5;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() throws IOException {
+ int numSubpartitions = 10;
+ int numBuffersInPool = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageMemoryManagerImpl storageMemoryManager =
+ createStorageMemoryManager(numBuffersInPool);
+ BufferAccumulator bufferAccumulator =
+ new HashBufferAccumulator(numSubpartitions, bufferSize, storageMemoryManager);
Review Comment:
Because we have mocked a `TesttingBufferAccumulatr`, we don't need this anymore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22652: [FLINK-31640][network] Introduce the segment id tracker for the tiered storage
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22652:
URL: https://github.com/apache/flink/pull/22652#issuecomment-1562406744
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "96330571b839e8f056116f6c258cd5b04072e40b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "96330571b839e8f056116f6c258cd5b04072e40b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 96330571b839e8f056116f6c258cd5b04072e40b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214223911
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
+
+ private Supplier<Boolean[]> tryWriteReturnValueSupplier;
+
+ private int numTotalReceivedBuffers;
+
+ private int tryWriteCounter;
+
+ private boolean isClosed;
+
+ public void setTryStartNewSegmentReturnValueSupplier(
+ Supplier<Boolean> tryStartNewSegmentReturnValueSupplier) {
+ this.tryStartNewSegmentReturnValueSupplier = tryStartNewSegmentReturnValueSupplier;
+ }
+
+ public void setTryWriteReturnValueSupplier(Supplier<Boolean[]> tryWriteReturnValueSupplier) {
+ this.tryWriteReturnValueSupplier = tryWriteReturnValueSupplier;
+ }
+
+ public int numTotalReceivedBuffers() {
+ return numTotalReceivedBuffers;
+ }
+
+ public boolean isClosed() {
+ return isClosed;
+ }
+
+ @Override
+ public boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId) {
+ return tryStartNewSegmentReturnValueSupplier.get();
+ }
+
+ @Override
+ public boolean tryWrite(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer) {
+ boolean isSuccess = tryWriteReturnValueSupplier.get()[(tryWriteCounter++) % 2];
+ if (isSuccess) {
+ numTotalReceivedBuffers++;
+ }
+ return isSuccess;
Review Comment:
Fixed.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214223224
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212793674
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -80,13 +107,23 @@ public void write(
if (isBroadcast && !isBroadcastOnly) {
for (int i = 0; i < numSubpartitions; ++i) {
- bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ // As the tiered storage subpartition ID is created only for broadcast records,
+ // which are fewer than normal records, the performance impact is expected to be
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214031240
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -80,13 +107,23 @@ public void write(
if (isBroadcast && !isBroadcastOnly) {
for (int i = 0; i < numSubpartitions; ++i) {
- bufferAccumulator.receive(record.duplicate(), subpartitionId, dataType);
+ // As the tiered storage subpartition ID is created only for broadcast records,
+ // which are fewer than normal records, the performance impact of generating new
+ // TieredStorageSubpartitionId objects is expected to be manageable. If the
+ // performance is significantly affected, this logic will be optimized accordingly.
+ bufferAccumulator.receive(
+ record.duplicate(), new TieredStorageSubpartitionId(i), dataType);
}
} else {
bufferAccumulator.receive(record, subpartitionId, dataType);
}
}
+ public void setMetricStatisticsUpdater(
+ Consumer<TieredStorageProducerMetricUpdate> metricStatisticsUpdater) {
+ this.metricStatisticsUpdater = metricStatisticsUpdater;
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1212695497
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +149,60 @@ private void writeAccumulatedBuffers(
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.
+ updateStatistics(accumulatedBuffer);
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+
+ if (currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()] == null) {
+ chooseStorageTierToStartSegment(subpartitionId);
+ }
+
+ boolean isSuccess =
+ currentSubpartitionTierAgent[subpartitionId.getSubpartitionId()].write(
Review Comment:
Added docs for the `write` and added docs for `TierProducerAgent` API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1214057806
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
Review Comment:
```suggestion
private Supplier<Boolean> tryStartNewSegmentSupplier;
```
```suggestion
private Supplier<Boolean> tryStartNewSegmentSupplier;
```
In general, for mock class, we named it's fields as follows:
`MethodName+(Runnable/Supplier/Consumer/Function)`
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
+
+ private Supplier<Boolean[]> tryWriteReturnValueSupplier;
Review Comment:
```suggestion
private BiFunction<TieredStorageSubpartitionId, Buffer, Boolean> tryWriterFunction;
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingTierProducerAgent;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.TieredStorageTestUtils.generateRandomData;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(numSubpartitions, Collections.emptyList());
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testWriteRecords() throws IOException {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ int maxNumToWriteRecordsPerSubpartition = 1000;
+ Random random = new Random();
+
+ TestingTierProducerAgent tierProducerAgent = new TestingTierProducerAgent();
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> true);
+ tierProducerAgent.setTryWriteReturnValueSupplier(() -> new Boolean[] {false, true});
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(
+ numSubpartitions, Collections.singletonList(tierProducerAgent));
+
+ int numWriteRecords = 0;
+ for (int j = 0; j < numSubpartitions; j++) {
+ for (int i = 0; i < random.nextInt(maxNumToWriteRecordsPerSubpartition); i++) {
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(j),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast);
+ numWriteRecords++;
+ }
+ }
+
+ int numExpectedBuffers = isBroadcast ? numWriteRecords * numSubpartitions : numWriteRecords;
+ assertThat(tierProducerAgent.numTotalReceivedBuffers()).isEqualTo(numExpectedBuffers);
+ }
+
+ @TestTemplate
+ void testTierCanNotStartNewSegment() throws IOException {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TestingTierProducerAgent tierProducerAgent = new TestingTierProducerAgent();
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> false);
+ tierProducerAgent.setTryWriteReturnValueSupplier(() -> new Boolean[] {false, true});
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(
+ numSubpartitions, Collections.singletonList(tierProducerAgent));
+
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> true);
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast);
+
+ assertThat(tierProducerAgent.numTotalReceivedBuffers())
+ .isEqualTo(isBroadcast ? numSubpartitions : 1);
+ }
+
+ @TestTemplate
+ void testClose() {
+ int numSubpartitions = 10;
+
+ TestingTierProducerAgent tierProducerAgent = new TestingTierProducerAgent();
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> true);
+ tierProducerAgent.setTryWriteReturnValueSupplier(() -> new Boolean[] {false, true});
Review Comment:
Are these two lines necessary? Because we haven't even call the `write` method, why do we care about them?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingTierProducerAgent;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.TieredStorageTestUtils.generateRandomData;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
Review Comment:
We also need to test metric updating in a separate test method.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
+
+ private Supplier<Boolean[]> tryWriteReturnValueSupplier;
+
+ private int numTotalReceivedBuffers;
+
+ private int tryWriteCounter;
+
+ private boolean isClosed;
+
+ public void setTryStartNewSegmentReturnValueSupplier(
+ Supplier<Boolean> tryStartNewSegmentReturnValueSupplier) {
+ this.tryStartNewSegmentReturnValueSupplier = tryStartNewSegmentReturnValueSupplier;
+ }
+
+ public void setTryWriteReturnValueSupplier(Supplier<Boolean[]> tryWriteReturnValueSupplier) {
+ this.tryWriteReturnValueSupplier = tryWriteReturnValueSupplier;
+ }
+
+ public int numTotalReceivedBuffers() {
+ return numTotalReceivedBuffers;
+ }
+
+ public boolean isClosed() {
+ return isClosed;
+ }
+
+ @Override
+ public boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId) {
+ return tryStartNewSegmentReturnValueSupplier.get();
+ }
+
+ @Override
+ public boolean tryWrite(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer) {
+ boolean isSuccess = tryWriteReturnValueSupplier.get()[(tryWriteCounter++) % 2];
+ if (isSuccess) {
+ numTotalReceivedBuffers++;
+ }
+ return isSuccess;
Review Comment:
Control logic should be put in specific testing method instead of mock class.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
Review Comment:
For testing(mock) class, we usually via `builder` pattern to make it's fields immutable. You can refer to `TestingSlotAllocator` or `TestingHsDataView`.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/TestingTierProducerAgent.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.tier.TierProducerAgent;
+
+import java.util.function.Supplier;
+
+/** Test implementation for {@link TierProducerAgent}. */
+public class TestingTierProducerAgent implements TierProducerAgent {
+
+ private Supplier<Boolean> tryStartNewSegmentReturnValueSupplier;
+
+ private Supplier<Boolean[]> tryWriteReturnValueSupplier;
+
+ private int numTotalReceivedBuffers;
+
+ private int tryWriteCounter;
+
+ private boolean isClosed;
+
+ public void setTryStartNewSegmentReturnValueSupplier(
+ Supplier<Boolean> tryStartNewSegmentReturnValueSupplier) {
+ this.tryStartNewSegmentReturnValueSupplier = tryStartNewSegmentReturnValueSupplier;
+ }
+
+ public void setTryWriteReturnValueSupplier(Supplier<Boolean[]> tryWriteReturnValueSupplier) {
+ this.tryWriteReturnValueSupplier = tryWriteReturnValueSupplier;
+ }
+
+ public int numTotalReceivedBuffers() {
+ return numTotalReceivedBuffers;
+ }
+
+ public boolean isClosed() {
+ return isClosed;
+ }
+
+ @Override
+ public boolean tryStartNewSegment(TieredStorageSubpartitionId subpartitionId, int segmentId) {
+ return tryStartNewSegmentReturnValueSupplier.get();
+ }
+
+ @Override
+ public boolean tryWrite(TieredStorageSubpartitionId subpartitionId, Buffer finishedBuffer) {
+ boolean isSuccess = tryWriteReturnValueSupplier.get()[(tryWriteCounter++) % 2];
+ if (isSuccess) {
+ numTotalReceivedBuffers++;
+ }
+ return isSuccess;
+ }
+
+ @Override
+ public void close() {
+ this.isClosed = true;
Review Comment:
It is best to use `closeRunnable()` as it will be more versatile. Mock testing class should be flexible and reusable.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClientTest.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.NetworkBufferPool;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingBufferAccumulator;
+import org.apache.flink.runtime.io.network.partition.hybrid.tiered.TestingTierProducerAgent;
+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.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.flink.runtime.io.network.partition.hybrid.tiered.TieredStorageTestUtils.generateRandomData;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/** Tests for {@link TieredStorageProducerClient}. */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TieredStorageProducerClientTest {
+
+ private static final int NUM_TOTAL_BUFFERS = 1000;
+
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ @Parameter public boolean isBroadcast;
+
+ private NetworkBufferPool globalPool;
+
+ @Parameters(name = "isBroadcast={0}")
+ public static Collection<Boolean> parameters() {
+ return Arrays.asList(false, true);
+ }
+
+ @BeforeEach
+ void before() {
+ globalPool = new NetworkBufferPool(NUM_TOTAL_BUFFERS, NETWORK_BUFFER_SIZE);
+ }
+
+ @AfterEach
+ void after() {
+ globalPool.destroy();
+ }
+
+ @TestTemplate
+ void testWriteRecordsToEmptyStorageTiers() {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ Random random = new Random();
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(numSubpartitions, Collections.emptyList());
+ assertThatThrownBy(
+ () ->
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(0),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to choose a storage tier");
+ }
+
+ @TestTemplate
+ void testWriteRecords() throws IOException {
+ int numSubpartitions = 10;
+ int bufferSize = 1024;
+ int maxNumToWriteRecordsPerSubpartition = 1000;
+ Random random = new Random();
+
+ TestingTierProducerAgent tierProducerAgent = new TestingTierProducerAgent();
+ tierProducerAgent.setTryStartNewSegmentReturnValueSupplier(() -> true);
+ tierProducerAgent.setTryWriteReturnValueSupplier(() -> new Boolean[] {false, true});
+
+ TieredStorageProducerClient tieredStorageProducerClient =
+ createTieredStorageProducerClient(
+ numSubpartitions, Collections.singletonList(tierProducerAgent));
+
+ int numWriteRecords = 0;
+ for (int j = 0; j < numSubpartitions; j++) {
+ for (int i = 0; i < random.nextInt(maxNumToWriteRecordsPerSubpartition); i++) {
+ tieredStorageProducerClient.write(
+ generateRandomData(bufferSize, random),
+ new TieredStorageSubpartitionId(j),
+ Buffer.DataType.DATA_BUFFER,
+ isBroadcast);
+ numWriteRecords++;
+ }
+ }
+
+ int numExpectedBuffers = isBroadcast ? numWriteRecords * numSubpartitions : numWriteRecords;
Review Comment:
I wonder if repeating write multiple times here is actually equivalent to `tryWrite (false) -> tryStartNewSegment -> tryWrite (true)`? 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #22652: [FLINK-31640][network] Write the accumulated buffers to the right storage tier
Posted by "TanYuxin-tyx (via GitHub)" <gi...@apache.org>.
TanYuxin-tyx commented on code in PR #22652:
URL: https://github.com/apache/flink/pull/22652#discussion_r1211224068
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/storage/TieredStorageProducerClient.java:
##########
@@ -119,7 +145,66 @@ private void writeAccumulatedBuffers(
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.
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int tierIndex = chooseStorageTierIfNeeded(subpartitionId);
+
+ Buffer compressedBuffer = compressBufferIfPossible(accumulatedBuffer);
+ updateStatistics(compressedBuffer);
+ boolean isLastBufferInSegment =
+ tierProducerAgents.get(tierIndex).write(subpartitionId, compressedBuffer);
+ if (isLastBufferInSegment) {
+ isSubpartitionSegmentFinished[subpartitionIndex] = true;
+ }
+ }
+
+ private int chooseStorageTierIfNeeded(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ if (isSubpartitionSegmentFinished[subpartitionIndex]) {
+ // If the current segment is finished, this subpartition need choose a new storage tier.
+ internalChooseStorageTier(subpartitionId);
+ isSubpartitionSegmentFinished[subpartitionIndex] = false;
+ }
+ return currentSubpartitionTierIndex[subpartitionIndex];
+ }
+
+ private void internalChooseStorageTier(TieredStorageSubpartitionId subpartitionId)
+ throws IOException {
+ int subpartitionIndex = subpartitionId.getSubpartitionId();
+ int segmentIndex = currentSubpartitionSegmentId[subpartitionIndex];
+ int nextSegmentIndex = segmentIndex + 1;
+
+ for (int tierIndex = 0; tierIndex < tierProducerAgents.size(); ++tierIndex) {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org