You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sx...@apache.org on 2024/09/20 11:48:40 UTC
(flink) 03/04: [FLINK-33750][config] Remove floating/exclusive buffer config options
This is an automated email from the ASF dual-hosted git repository.
sxnan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ef40181163859311ab0e757d632109675e2c310d
Author: sxnan <su...@gmail.com>
AuthorDate: Wed Sep 18 23:43:39 2024 +0800
[FLINK-33750][config] Remove floating/exclusive buffer config options
---
.../NettyShuffleEnvironmentOptions.java | 85 ----------------------
.../network/partition/PipelinedSubpartition.java | 5 +-
.../flink/runtime/shuffle/NettyShuffleMaster.java | 19 ++---
.../NettyShuffleEnvironmentConfiguration.java | 15 +---
.../io/network/NettyShuffleEnvironmentTest.java | 11 +--
.../NettyShuffleEnvironmentConfigurationTest.java | 8 +-
.../StreamNetworkThroughputBenchmarkTest.java | 21 +-----
.../test/checkpointing/AutoRescalingITCase.java | 17 ++---
.../EventTimeWindowCheckpointingITCase.java | 12 +--
.../test/checkpointing/LocalRecoveryITCase.java | 2 +-
.../flink/test/checkpointing/RescalingITCase.java | 15 +---
.../checkpointing/UnalignedCheckpointITCase.java | 19 +----
.../UnalignedCheckpointRescaleITCase.java | 19 +----
.../checkpointing/UnalignedCheckpointTestBase.java | 8 --
14 files changed, 35 insertions(+), 221 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
index 07ebce9d1ce..7d516fbe3ee 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java
@@ -149,72 +149,6 @@ public class NettyShuffleEnvironmentOptions {
+ " and 1000 for batch workloads. If explicitly configured, the"
+ " configured value should be at least 1.");
- /**
- * Number of network buffers for each outgoing/incoming channel (subpartition/input channel).
- * The minimum valid value for the option is 0. When the option is configured as 0, the
- * exclusive network buffers used per downstream incoming channel will be 0, but for each
- * upstream outgoing channel, max(1, configured value) will be used. In other words we ensure
- * that, for performance reasons, at least one buffer is used per outgoing channel regardless of
- * the configuration.
- *
- * @deprecated This option is deprecated in 1.20 and will be removed in 2.0 to simplify the
- * configuration of network buffers.
- */
- @Deprecated
- public static final ConfigOption<Integer> NETWORK_BUFFERS_PER_CHANNEL =
- key("taskmanager.network.memory.buffers-per-channel")
- .intType()
- .defaultValue(2)
- .withDescription(
- String.format(
- "Number of exclusive network buffers for each outgoing/incoming"
- + " channel (subpartition/input channel) in the credit-based"
- + " flow control model. For the outgoing channel(subpartition),"
- + " this value is the effective exclusive buffers per channel."
- + " For the incoming channel(input channel), this value"
- + " is the max number of exclusive buffers per channel,"
- + " the number of effective exclusive network buffers per"
- + " channel is dynamically calculated from %s and the"
- + " effective range is from 0 to the configured value."
- + " The minimum valid value for the option is 0. When"
- + " the option is configured as 0, the exclusive network"
- + " buffers used by downstream incoming channel will be"
- + " 0, but for each upstream outgoing channel, max(1,"
- + " configured value) will be used. In other words, we"
- + " ensure that, for performance reasons, at least one"
- + " buffer is used per outgoing channel regardless of"
- + " the configuration.",
- NETWORK_READ_MAX_REQUIRED_BUFFERS_PER_GATE.key()));
-
- /**
- * Number of floating network buffers for each outgoing/incoming gate (result partition/input
- * gate).
- *
- * @deprecated This option is deprecated in 1.20 and will be removed in 2.0 to simplify the
- * configuration of network buffers.
- */
- @Deprecated
- public static final ConfigOption<Integer> NETWORK_EXTRA_BUFFERS_PER_GATE =
- key("taskmanager.network.memory.floating-buffers-per-gate")
- .intType()
- .defaultValue(8)
- .withDescription(
- String.format(
- "Number of floating network buffers for each outgoing/incoming"
- + " gate (result partition/input gate). In credit-based"
- + " flow control mode, this indicates how many floating"
- + " credits are shared among all the channels. The floating"
- + " buffers can help relieve back-pressure caused by"
- + " unbalanced data distribution among the subpartitions."
- + " For the outgoing gate(result partition), this value"
- + " is the effective floating buffers per gate. For the"
- + " incoming gate(input gate), this value is a recommended"
- + " number of floating buffers, the number of effective"
- + " floating network buffers per gate is dynamically"
- + " calculated from %s and the range of effective floating"
- + " buffers is from 0 to (parallelism - 1).",
- NETWORK_READ_MAX_REQUIRED_BUFFERS_PER_GATE.key()));
-
/**
* Minimum number of network buffers required per blocking result partition for sort-shuffle.
*/
@@ -263,25 +197,6 @@ public class NettyShuffleEnvironmentOptions {
// this raw value must be changed correspondingly
"taskmanager.memory.framework.off-heap.batch-shuffle.size"));
- /**
- * Number of max buffers can be used for each output subpartition.
- *
- * @deprecated This option is deprecated in 1.20 and will be removed in 2.0 to simplify the
- * configuration of network buffers.
- */
- @Deprecated
- public static final ConfigOption<Integer> NETWORK_MAX_BUFFERS_PER_CHANNEL =
- key("taskmanager.network.memory.max-buffers-per-channel")
- .intType()
- .defaultValue(10)
- .withDescription(
- "Number of max buffers that can be used for each channel. If a channel exceeds the number of max"
- + " buffers, it will make the task become unavailable, cause the back pressure and block the data processing. This"
- + " might speed up checkpoint alignment by preventing excessive growth of the buffered in-flight data in"
- + " case of data skew and high number of configured floating buffers. This limit is not strictly guaranteed,"
- + " and can be ignored by things like flatMap operators, records spanning multiple buffers or single timer"
- + " producing large amount of data.");
-
/** The timeout for requesting buffers for each channel. */
@Documentation.ExcludeFromDocumentation(
"This option is purely implementation related, and may be removed as the implementation changes.")
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
index d3e4caf42ba..a093ab9313a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java
@@ -75,10 +75,7 @@ public class PipelinedSubpartition extends ResultSubpartition implements Channel
// ------------------------------------------------------------------------
- /**
- * Number of exclusive credits per input channel at the downstream tasks configured by {@link
- * org.apache.flink.configuration.NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_PER_CHANNEL}.
- */
+ /** Number of exclusive credits per input channel at the downstream tasks. */
private final int receiverExclusiveBuffersPerChannel;
/** All buffers of this subpartition. Access to the buffers is synchronized on this object. */
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
index 66191d72071..791d5022b35 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/shuffle/NettyShuffleMaster.java
@@ -69,10 +69,8 @@ public class NettyShuffleMaster implements ShuffleMaster<NettyShuffleDescriptor>
public NettyShuffleMaster(ShuffleMasterContext shuffleMasterContext) {
Configuration conf = shuffleMasterContext.getConfiguration();
checkNotNull(conf);
- buffersPerInputChannel =
- conf.get(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL);
- floatingBuffersPerGate =
- conf.get(NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE);
+ buffersPerInputChannel = 2;
+ floatingBuffersPerGate = 8;
maxRequiredBuffersPerGate =
conf.getOptional(
NettyShuffleEnvironmentOptions.NETWORK_READ_MAX_REQUIRED_BUFFERS_PER_GATE);
@@ -94,11 +92,6 @@ public class NettyShuffleMaster implements ShuffleMaster<NettyShuffleDescriptor>
"At least one buffer is required for each gate, please increase the value of %s.",
NettyShuffleEnvironmentOptions.NETWORK_READ_MAX_REQUIRED_BUFFERS_PER_GATE
.key()));
- checkArgument(
- floatingBuffersPerGate >= 1,
- String.format(
- "The configured floating buffer should be at least 1, please increase the value of %s.",
- NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE.key()));
}
@Override
@@ -147,11 +140,9 @@ public class NettyShuffleMaster implements ShuffleMaster<NettyShuffleDescriptor>
/**
* JM announces network memory requirement from the calculating result of this method. Please
* note that the calculating algorithm depends on both I/O details of a vertex and network
- * configuration, e.g. {@link NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_PER_CHANNEL} and
- * {@link NettyShuffleEnvironmentOptions#NETWORK_EXTRA_BUFFERS_PER_GATE}, which means we should
- * always keep the consistency of configurations between JM, RM and TM in fine-grained resource
- * management, thus to guarantee that the processes of memory announcing and allocating respect
- * each other.
+ * configuration, which means we should always keep the consistency of configurations between
+ * JM, RM and TM in fine-grained resource management, thus to guarantee that the processes of
+ * memory announcing and allocating respect each other.
*/
@Override
public MemorySize computeShuffleMemorySizeForTask(TaskInputsOutputsDescriptor desc) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
index f2005fce73c..b873e486504 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java
@@ -304,17 +304,14 @@ public class NettyShuffleEnvironmentConfiguration {
.NETWORK_PARTITION_REQUEST_TIMEOUT)
.toMillis();
- int buffersPerChannel =
- configuration.get(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL);
- int extraBuffersPerGate =
- configuration.get(NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE);
+ int buffersPerChannel = 2;
+ int extraBuffersPerGate = 8;
Optional<Integer> maxRequiredBuffersPerGate =
configuration.getOptional(
NettyShuffleEnvironmentOptions.NETWORK_READ_MAX_REQUIRED_BUFFERS_PER_GATE);
- int maxBuffersPerChannel =
- configuration.get(NettyShuffleEnvironmentOptions.NETWORK_MAX_BUFFERS_PER_CHANNEL);
+ int maxBuffersPerChannel = 10;
long batchShuffleReadMemoryBytes =
configuration.get(TaskManagerOptions.NETWORK_BATCH_SHUFFLE_READ_MEMORY).getBytes();
@@ -349,18 +346,12 @@ public class NettyShuffleEnvironmentConfiguration {
configuration.get(
NettyShuffleEnvironmentOptions.TCP_CONNECTION_REUSE_ACROSS_JOBS_ENABLED);
- checkArgument(buffersPerChannel >= 0, "Must be non-negative.");
checkArgument(
!maxRequiredBuffersPerGate.isPresent() || maxRequiredBuffersPerGate.get() >= 1,
String.format(
"At least one buffer is required for each gate, please increase the value of %s.",
NettyShuffleEnvironmentOptions.NETWORK_READ_MAX_REQUIRED_BUFFERS_PER_GATE
.key()));
- checkArgument(
- extraBuffersPerGate >= 1,
- String.format(
- "The configured floating buffer should be at least 1, please increase the value of %s.",
- NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE.key()));
TieredStorageConfiguration tieredStorageConfiguration = null;
if ((configuration.get(BATCH_SHUFFLE_MODE) == ALL_EXCHANGES_HYBRID_FULL
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
index ae792398de4..c626a2c7336 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NettyShuffleEnvironmentTest.java
@@ -20,7 +20,6 @@ package org.apache.flink.runtime.io.network;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.testutils.BlockerSync;
import org.apache.flink.metrics.CharacterFilter;
@@ -104,8 +103,7 @@ class NettyShuffleEnvironmentTest {
void testRegisterTaskWithLimitedBuffers() throws Exception {
// outgoing: 1 buffer per channel + 1 extra buffer per ResultPartition
// incoming: 2 exclusive buffers per channel + 1 floating buffer per single gate
- final int bufferCount =
- 18 + 10 * NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL.defaultValue();
+ final int bufferCount = 18 + 10 * 2;
testRegisterTaskWithLimitedBuffers(bufferCount);
}
@@ -118,12 +116,7 @@ class NettyShuffleEnvironmentTest {
void testRegisterTaskWithInsufficientBuffers() throws Exception {
// outgoing: 1 buffer per channel + 1 extra buffer per ResultPartition
// incoming: 2 exclusive buffers per channel + 1 floating buffer per single gate
- final int bufferCount =
- 10
- + 10
- * NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL
- .defaultValue()
- - 1;
+ final int bufferCount = 10 + 10 * 2 - 1;
assertThatThrownBy(() -> testRegisterTaskWithLimitedBuffers(bufferCount))
.isInstanceOf(IOException.class)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NettyShuffleEnvironmentConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NettyShuffleEnvironmentConfigurationTest.java
index 89e4b998b50..ce15a459772 100755
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NettyShuffleEnvironmentConfigurationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/NettyShuffleEnvironmentConfigurationTest.java
@@ -53,9 +53,7 @@ class NettyShuffleEnvironmentConfigurationTest {
* Verifies that {@link NettyShuffleEnvironmentConfiguration#fromConfiguration(Configuration,
* MemorySize, boolean, InetAddress)} returns the correct result for new configurations via
* {@link NettyShuffleEnvironmentOptions#NETWORK_REQUEST_BACKOFF_INITIAL}, {@link
- * NettyShuffleEnvironmentOptions#NETWORK_REQUEST_BACKOFF_MAX}, {@link
- * NettyShuffleEnvironmentOptions#NETWORK_BUFFERS_PER_CHANNEL} and {@link
- * NettyShuffleEnvironmentOptions#NETWORK_EXTRA_BUFFERS_PER_GATE}
+ * NettyShuffleEnvironmentOptions#NETWORK_REQUEST_BACKOFF_MAX}
*/
@Test
void testNetworkRequestBackoffAndBuffers() {
@@ -64,8 +62,6 @@ class NettyShuffleEnvironmentConfigurationTest {
final Configuration config = new Configuration();
config.set(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100);
config.set(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 200);
- config.set(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL, 10);
- config.set(NettyShuffleEnvironmentOptions.NETWORK_EXTRA_BUFFERS_PER_GATE, 100);
final NettyShuffleEnvironmentConfiguration networkConfig =
NettyShuffleEnvironmentConfiguration.fromConfiguration(
@@ -73,8 +69,6 @@ class NettyShuffleEnvironmentConfigurationTest {
assertThat(networkConfig.partitionRequestInitialBackoff()).isEqualTo(100);
assertThat(networkConfig.partitionRequestMaxBackoff()).isEqualTo(200);
- assertThat(networkConfig.networkBuffersPerChannel()).isEqualTo(10);
- assertThat(networkConfig.floatingNetworkBuffersPerGate()).isEqualTo(100);
}
/** Verifies the correlation of sort-merge blocking shuffle config options. */
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTest.java
index e9011fe51dd..6e34751ec84 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/benchmark/StreamNetworkThroughputBenchmarkTest.java
@@ -18,8 +18,6 @@
package org.apache.flink.streaming.runtime.io.benchmark;
-import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
-
import org.junit.jupiter.api.Test;
import java.io.IOException;
@@ -82,11 +80,7 @@ class StreamNetworkThroughputBenchmarkTest {
100,
false,
writers * channels - 1,
- writers
- * channels
- * NettyShuffleEnvironmentOptions
- .NETWORK_BUFFERS_PER_CHANNEL
- .defaultValue()))
+ writers * channels * 2))
.isInstanceOf(IOException.class)
.hasMessageContaining("Insufficient number of network buffers");
}
@@ -105,12 +99,7 @@ class StreamNetworkThroughputBenchmarkTest {
100,
false,
writers * channels,
- writers
- * channels
- * NettyShuffleEnvironmentOptions
- .NETWORK_BUFFERS_PER_CHANNEL
- .defaultValue()
- - 1))
+ writers * channels * 2 - 1))
.isInstanceOf(IOException.class)
.hasMessageContaining("Insufficient number of network buffers");
}
@@ -127,11 +116,7 @@ class StreamNetworkThroughputBenchmarkTest {
100,
false,
writers * channels + writers,
- writers
- + writers
- * channels
- * NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL
- .defaultValue());
+ writers + writers * channels * 2);
env.executeBenchmark(10_000);
env.tearDown();
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java
index b1f1cd7b7d3..d3dfcc951a9 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/AutoRescalingITCase.java
@@ -34,7 +34,6 @@ import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.configuration.StateRecoveryOptions;
import org.apache.flink.configuration.WebOptions;
@@ -111,27 +110,23 @@ public class AutoRescalingITCase extends TestLogger {
private static final int slotsPerTaskManager = 2;
private static final int totalSlots = numTaskManagers * slotsPerTaskManager;
- @Parameterized.Parameters(name = "backend = {0}, buffersPerChannel = {1}, useIngestDB = {2}")
+ @Parameterized.Parameters(name = "backend = {0}, useIngestDB = {1}")
public static Collection<Object[]> data() {
return Arrays.asList(
new Object[][] {
- {"rocksdb", 0, false},
- {"rocksdb", 2, true},
- {"filesystem", 0, false},
- {"filesystem", 2, false}
+ {"rocksdb", false},
+ {"rocksdb", true},
+ {"filesystem", false}
});
}
- public AutoRescalingITCase(String backend, int buffersPerChannel, boolean useIngestDB) {
+ public AutoRescalingITCase(String backend, boolean useIngestDB) {
this.backend = backend;
- this.buffersPerChannel = buffersPerChannel;
this.useIngestDB = useIngestDB;
}
private final String backend;
- private final int buffersPerChannel;
-
private final boolean useIngestDB;
private String currentBackend = null;
@@ -168,8 +163,6 @@ public class AutoRescalingITCase extends TestLogger {
config.set(
CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir.toURI().toString());
config.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointDir.toURI().toString());
- config.set(
- NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL, buffersPerChannel);
config.set(JobManagerOptions.SCHEDULER, JobManagerOptions.SchedulerType.Adaptive);
// Disable the scaling cooldown to speed up the test
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
index d2606dc308a..0e494973802 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
@@ -29,7 +29,6 @@ import org.apache.flink.changelog.fs.FsStateChangelogStorageFactory;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.RpcOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
@@ -109,8 +108,6 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
public StateBackendEnum stateBackendEnum;
- private final int buffersPerChannel;
-
enum StateBackendEnum {
MEM,
FILE,
@@ -119,18 +116,16 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
ROCKSDB_INCREMENTAL_ZK,
}
- @Parameterized.Parameters(name = "statebackend type ={0}, buffersPerChannel = {1}")
+ @Parameterized.Parameters(name = "statebackend type ={0}")
public static Collection<Object[]> parameter() {
return Arrays.stream(StateBackendEnum.values())
- .map((type) -> new Object[][] {{type, 0}, {type, 2}})
+ .map((type) -> new Object[][] {{type}})
.flatMap(Arrays::stream)
.collect(Collectors.toList());
}
- public EventTimeWindowCheckpointingITCase(
- StateBackendEnum stateBackendEnum, int buffersPerChannel) {
+ public EventTimeWindowCheckpointingITCase(StateBackendEnum stateBackendEnum) {
this.stateBackendEnum = stateBackendEnum;
- this.buffersPerChannel = buffersPerChannel;
}
protected StateBackendEnum getStateBackend() {
@@ -168,7 +163,6 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
}
Configuration config = createClusterConfig();
- config.set(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL, buffersPerChannel);
switch (stateBackendEnum) {
case MEM:
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/LocalRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/LocalRecoveryITCase.java
index 6442031e111..e991ff51d73 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/LocalRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/LocalRecoveryITCase.java
@@ -132,7 +132,7 @@ public class LocalRecoveryITCase extends TestLogger {
StateBackendEnum backendEnum,
boolean localBackupEnable,
boolean localRecoveryEnabled) {
- super(backendEnum, 2);
+ super(backendEnum);
this.backendEnum = backendEnum;
this.localBackupEnable = localBackupEnable;
this.localRecoveryEnabled = localRecoveryEnabled;
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
index 6d2bde2b3be..3d746d229fc 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
@@ -31,7 +31,6 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.client.JobExecutionException;
@@ -102,23 +101,17 @@ public class RescalingITCase extends TestLogger {
private static final int slotsPerTaskManager = 2;
private static final int numSlots = numTaskManagers * slotsPerTaskManager;
- @Parameterized.Parameters(name = "backend = {0}, buffersPerChannel = {1}")
+ @Parameterized.Parameters(name = "backend = {0}")
public static Collection<Object[]> data() {
- return Arrays.asList(
- new Object[][] {
- {"filesystem", 2}, {"rocksdb", 0}, {"filesystem", 0}, {"rocksdb", 2}
- });
+ return Arrays.asList(new Object[][] {{"filesystem"}, {"rocksdb"}});
}
- public RescalingITCase(String backend, int buffersPerChannel) {
+ public RescalingITCase(String backend) {
this.backend = backend;
- this.buffersPerChannel = buffersPerChannel;
}
private final String backend;
- private final int buffersPerChannel;
-
private String currentBackend = null;
enum OperatorCheckpointMethod {
@@ -149,8 +142,6 @@ public class RescalingITCase extends TestLogger {
config.set(
CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir.toURI().toString());
config.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointDir.toURI().toString());
- config.set(
- NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL, buffersPerChannel);
cluster =
new MiniClusterWithClientResource(
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
index 2fa298c9984..5f167cc58b7 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
@@ -228,8 +228,7 @@ public class UnalignedCheckpointITCase extends UnalignedCheckpointTestBase {
}
}
- @Parameterized.Parameters(
- name = "{0} with {2} channels, p = {1}, timeout = {3}, buffersPerChannel = {4}")
+ @Parameterized.Parameters(name = "{0} with {2} channels, p = {1}, timeout = {3}")
public static Object[][] parameters() {
Object[] defaults = {Topology.PIPELINE, 1, MIXED, 0};
@@ -248,12 +247,7 @@ public class UnalignedCheckpointITCase extends UnalignedCheckpointTestBase {
};
return Stream.of(runs)
.map(params -> addDefaults(params, defaults))
- .map(
- params ->
- new Object[][] {
- ArrayUtils.insert(params.length, params, 0),
- ArrayUtils.insert(params.length, params, BUFFER_PER_CHANNEL)
- })
+ .map(params -> new Object[][] {ArrayUtils.insert(params.length, params)})
.flatMap(Arrays::stream)
.toArray(Object[][]::new);
}
@@ -266,11 +260,7 @@ public class UnalignedCheckpointITCase extends UnalignedCheckpointTestBase {
private final UnalignedSettings settings;
public UnalignedCheckpointITCase(
- Topology topology,
- int parallelism,
- ChannelType channelType,
- int timeout,
- int buffersPerChannel) {
+ Topology topology, int parallelism, ChannelType channelType, int timeout) {
settings =
new UnalignedSettings(topology)
.setParallelism(parallelism)
@@ -282,8 +272,7 @@ public class UnalignedCheckpointITCase extends UnalignedCheckpointTestBase {
// after triggering)
.setCheckpointTimeout(Duration.ofSeconds(30))
.setTolerableCheckpointFailures(3)
- .setAlignmentTimeout(timeout)
- .setBuffersPerChannel(buffersPerChannel);
+ .setAlignmentTimeout(timeout);
}
@Test
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleITCase.java
index 7a21cab7c1d..ecedf7ea7c6 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleITCase.java
@@ -70,7 +70,6 @@ public class UnalignedCheckpointRescaleITCase extends UnalignedCheckpointTestBas
private final Topology topology;
private final int oldParallelism;
private final int newParallelism;
- private final int buffersPerChannel;
private final long sourceSleepMs;
enum Topology implements DagCreator {
@@ -538,8 +537,7 @@ public class UnalignedCheckpointRescaleITCase extends UnalignedCheckpointTestBas
}
}
- @Parameterized.Parameters(
- name = "{0} {1} from {2} to {3}, sourceSleepMs = {4}, buffersPerChannel = {5}")
+ @Parameterized.Parameters(name = "{0} {1} from {2} to {3}, sourceSleepMs = {4}")
public static Object[][] getScaleFactors() {
// We use `sourceSleepMs` > 0 to test rescaling without backpressure and only very few
// captured in-flight records, see FLINK-31963.
@@ -596,12 +594,7 @@ public class UnalignedCheckpointRescaleITCase extends UnalignedCheckpointTestBas
new Object[] {"no scale", Topology.MULTI_INPUT, 7, 7, 0L},
};
return Arrays.stream(parameters)
- .map(
- params ->
- new Object[][] {
- ArrayUtils.insert(params.length, params, 0),
- ArrayUtils.insert(params.length, params, BUFFER_PER_CHANNEL)
- })
+ .map(params -> new Object[][] {ArrayUtils.insert(params.length, params)})
.flatMap(Arrays::stream)
.toArray(Object[][]::new);
}
@@ -611,13 +604,11 @@ public class UnalignedCheckpointRescaleITCase extends UnalignedCheckpointTestBas
Topology topology,
int oldParallelism,
int newParallelism,
- long sourceSleepMs,
- int buffersPerChannel) {
+ long sourceSleepMs) {
this.topology = topology;
this.oldParallelism = oldParallelism;
this.newParallelism = newParallelism;
this.sourceSleepMs = sourceSleepMs;
- this.buffersPerChannel = buffersPerChannel;
}
@Test
@@ -626,7 +617,6 @@ public class UnalignedCheckpointRescaleITCase extends UnalignedCheckpointTestBas
new UnalignedSettings(topology)
.setParallelism(oldParallelism)
.setExpectedFailures(1)
- .setBuffersPerChannel(buffersPerChannel)
.setSourceSleepMs(sourceSleepMs);
prescaleSettings.setGenerateCheckpoint(true);
final File checkpointDir = super.execute(prescaleSettings);
@@ -635,8 +625,7 @@ public class UnalignedCheckpointRescaleITCase extends UnalignedCheckpointTestBas
final UnalignedSettings postscaleSettings =
new UnalignedSettings(topology)
.setParallelism(newParallelism)
- .setExpectedFailures(1)
- .setBuffersPerChannel(buffersPerChannel);
+ .setExpectedFailures(1);
postscaleSettings.setRestoreCheckpoint(checkpointDir);
super.execute(postscaleSettings);
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java
index 83965b57a74..46cdb2bf19b 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java
@@ -120,7 +120,6 @@ public abstract class UnalignedCheckpointTestBase extends TestLogger {
protected static final String NUM_FAILURES = "failures";
protected static final String NUM_DUPLICATES = "duplicates";
protected static final String NUM_LOST = "lost";
- protected static final int BUFFER_PER_CHANNEL = 1;
/** For multi-gate tests. */
protected static final int NUM_SOURCES = 3;
@@ -684,7 +683,6 @@ public abstract class UnalignedCheckpointTestBase extends TestLogger {
CheckpointingOptions.CHECKPOINTING_TIMEOUT.defaultValue();
private int failuresAfterSourceFinishes = 0;
private ChannelType channelType = ChannelType.MIXED;
- private int buffersPerChannel = 1;
private long sourceSleepMs = 0;
public UnalignedSettings(DagCreator dagCreator) {
@@ -736,11 +734,6 @@ public abstract class UnalignedCheckpointTestBase extends TestLogger {
return this;
}
- public UnalignedSettings setBuffersPerChannel(int buffersPerChannel) {
- this.buffersPerChannel = buffersPerChannel;
- return this;
- }
-
public UnalignedSettings setSourceSleepMs(long sourceSleepMs) {
this.sourceSleepMs = sourceSleepMs;
return this;
@@ -779,7 +772,6 @@ public abstract class UnalignedCheckpointTestBase extends TestLogger {
conf.set(
ShuffleServiceOptions.SHUFFLE_SERVICE_FACTORY_CLASS,
SharedPoolNettyShuffleServiceFactory.class.getName());
- conf.set(NettyShuffleEnvironmentOptions.NETWORK_BUFFERS_PER_CHANNEL, buffersPerChannel);
conf.set(NettyShuffleEnvironmentOptions.NETWORK_REQUEST_BACKOFF_MAX, 60000);
// half memory consumption of network buffers (default is 64mb), as some tests spawn a
// large number of task managers (25)