You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/06/07 09:00:00 UTC
[flink] branch release-1.17 updated (aaf72e45b9b -> ff52b475d14)
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a change to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
from aaf72e45b9b [FLINK-28853][Docs] Sync watermark split alignment documentation with the Chinese version. This closes #22724
new c76c162835c [hotfix] Migrate SlotManagerConfigurationTest to Junit5 and AssertJ.
new ff52b475d14 [FLINK-32254][runtime] FineGrainedSlotManager may not allocate enough taskmanagers if maxSlotNum is configured
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../slotmanager/SlotManagerConfiguration.java | 13 +++--
.../slotmanager/SlotManagerConfigurationTest.java | 67 +++++++++++++++++-----
2 files changed, 62 insertions(+), 18 deletions(-)
[flink] 01/02: [hotfix] Migrate SlotManagerConfigurationTest to Junit5 and AssertJ.
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c76c162835c762451dc46824d2b42507d142bea8
Author: Weijie Guo <re...@163.com>
AuthorDate: Mon Jun 5 19:54:02 2023 +0800
[hotfix] Migrate SlotManagerConfigurationTest to Junit5 and AssertJ.
---
.../slotmanager/SlotManagerConfigurationTest.java | 23 +++++++++-------------
1 file changed, 9 insertions(+), 14 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfigurationTest.java
index c37ec035e1f..782392e6378 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfigurationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfigurationTest.java
@@ -22,23 +22,20 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.runtime.resourcemanager.WorkerResourceSpec;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link SlotManagerConfiguration}. */
-public class SlotManagerConfigurationTest extends TestLogger {
+class SlotManagerConfigurationTest {
/**
* Tests that {@link SlotManagerConfiguration#getSlotRequestTimeout()} returns the value
* configured under key {@link JobManagerOptions#SLOT_REQUEST_TIMEOUT}.
*/
@Test
- public void testSetSlotRequestTimeout() throws Exception {
+ void testSetSlotRequestTimeout() throws Exception {
final long slotIdleTimeout = 42;
final Configuration configuration = new Configuration();
@@ -46,9 +43,8 @@ public class SlotManagerConfigurationTest extends TestLogger {
final SlotManagerConfiguration slotManagerConfiguration =
SlotManagerConfiguration.fromConfiguration(configuration, WorkerResourceSpec.ZERO);
- assertThat(
- slotManagerConfiguration.getSlotRequestTimeout().toMilliseconds(),
- is(equalTo(slotIdleTimeout)));
+ assertThat(slotIdleTimeout)
+ .isEqualTo(slotManagerConfiguration.getSlotRequestTimeout().toMilliseconds());
}
/**
@@ -56,7 +52,7 @@ public class SlotManagerConfigurationTest extends TestLogger {
* JobManagerOptions#SLOT_REQUEST_TIMEOUT} if set.
*/
@Test
- public void testPreferLegacySlotRequestTimeout() throws Exception {
+ void testPreferLegacySlotRequestTimeout() throws Exception {
final long legacySlotIdleTimeout = 42;
final Configuration configuration = new Configuration();
@@ -65,8 +61,7 @@ public class SlotManagerConfigurationTest extends TestLogger {
final SlotManagerConfiguration slotManagerConfiguration =
SlotManagerConfiguration.fromConfiguration(configuration, WorkerResourceSpec.ZERO);
- assertThat(
- slotManagerConfiguration.getSlotRequestTimeout().toMilliseconds(),
- is(equalTo(legacySlotIdleTimeout)));
+ assertThat(legacySlotIdleTimeout)
+ .isEqualTo(slotManagerConfiguration.getSlotRequestTimeout().toMilliseconds());
}
}
[flink] 02/02: [FLINK-32254][runtime] FineGrainedSlotManager may not allocate enough taskmanagers if maxSlotNum is configured
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ff52b475d14abbc8b7e6dd1fc19500a18491e78d
Author: Weijie Guo <re...@163.com>
AuthorDate: Mon Jun 5 21:04:38 2023 +0800
[FLINK-32254][runtime] FineGrainedSlotManager may not allocate enough taskmanagers if maxSlotNum is configured
---
.../slotmanager/SlotManagerConfiguration.java | 13 +++++--
.../slotmanager/SlotManagerConfigurationTest.java | 44 ++++++++++++++++++++++
2 files changed, 53 insertions(+), 4 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfiguration.java
index 9ca61779f41..676a086671b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfiguration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfiguration.java
@@ -220,8 +220,8 @@ public class SlotManagerConfiguration {
? new CPUResource(Double.MAX_VALUE)
: defaultWorkerResourceSpec
.getCpuCores()
- .divide(defaultWorkerResourceSpec.getNumSlots())
- .multiply(maxSlotNum));
+ .multiply(maxSlotNum)
+ .divide(defaultWorkerResourceSpec.getNumSlots()));
}
private static MemorySize getMaxTotalMem(
@@ -236,7 +236,12 @@ public class SlotManagerConfiguration {
? MemorySize.MAX_VALUE
: defaultWorkerResourceSpec
.getTotalMemSize()
- .divide(defaultWorkerResourceSpec.getNumSlots())
- .multiply(maxSlotNum));
+ // In theory, there is a possibility of long
+ // overflow here. However, in actual scenarios, for
+ // a 1TB of TM memory and a very large number of
+ // maxSlotNum (e.g. 1_000_000), there is still no
+ // overflow.
+ .multiply(maxSlotNum)
+ .divide(defaultWorkerResourceSpec.getNumSlots()));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfigurationTest.java
index 782392e6378..136a48d2bad 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfigurationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfigurationTest.java
@@ -20,11 +20,14 @@ package org.apache.flink.runtime.resourcemanager.slotmanager;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.runtime.resourcemanager.WorkerResourceSpec;
import org.junit.jupiter.api.Test;
+import java.math.BigDecimal;
+
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link SlotManagerConfiguration}. */
@@ -64,4 +67,45 @@ class SlotManagerConfigurationTest {
assertThat(legacySlotIdleTimeout)
.isEqualTo(slotManagerConfiguration.getSlotRequestTimeout().toMilliseconds());
}
+
+ @Test
+ void testComputeMaxTotalCpu() throws Exception {
+ final Configuration configuration = new Configuration();
+ final int maxSlotNum = 9;
+ final int numSlots = 3;
+ final double cpuCores = 10;
+ configuration.set(ResourceManagerOptions.MAX_SLOT_NUM, maxSlotNum);
+ final SlotManagerConfiguration slotManagerConfiguration =
+ SlotManagerConfiguration.fromConfiguration(
+ configuration,
+ new WorkerResourceSpec.Builder()
+ .setNumSlots(numSlots)
+ .setCpuCores(cpuCores)
+ .build());
+ assertThat(slotManagerConfiguration.getMaxTotalCpu().getValue().doubleValue())
+ .isEqualTo(cpuCores * maxSlotNum / numSlots);
+ }
+
+ @Test
+ void testComputeMaxTotalMemory() throws Exception {
+ final Configuration configuration = new Configuration();
+ final int maxSlotNum = 1_000_000;
+ final int numSlots = 10;
+ final int totalTaskManagerMB =
+ MemorySize.parse("1", MemorySize.MemoryUnit.TERA_BYTES).getMebiBytes();
+ configuration.set(ResourceManagerOptions.MAX_SLOT_NUM, maxSlotNum);
+ final SlotManagerConfiguration slotManagerConfiguration =
+ SlotManagerConfiguration.fromConfiguration(
+ configuration,
+ new WorkerResourceSpec.Builder()
+ .setNumSlots(numSlots)
+ .setTaskHeapMemoryMB(totalTaskManagerMB)
+ .build());
+ assertThat(slotManagerConfiguration.getMaxTotalMem().getBytes())
+ .isEqualTo(
+ BigDecimal.valueOf(MemorySize.ofMebiBytes(totalTaskManagerMB).getBytes())
+ .multiply(BigDecimal.valueOf(maxSlotNum))
+ .divide(BigDecimal.valueOf(numSlots))
+ .longValue());
+ }
}