This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5760787958 [flink] Support managed memory configuration for writer
slot sharing groups (#10767)
5760787958 is described below
commit 576078795883ff215ec676d053fed74c7ad4f41a
Author: Ran Tao <[email protected]>
AuthorDate: Mon Oct 26 14:24:27 2026 +0800
[flink] Support managed memory configuration for writer slot sharing groups
(#10767)
---
docs/docs/flink/configuration.md | 30 +++
docs/generated/flink_connector_configuration.html | 6 +
.../flink/sink/cdc/FlinkCdcMultiTableSink.java | 29 ++-
.../sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java | 3 +
.../flink/sink/cdc/FlinkCdcMultiTableSinkTest.java | 66 ++++++
.../apache/paimon/flink/FlinkConnectorOptions.java | 11 +
.../org/apache/paimon/flink/sink/FlinkSink.java | 36 +++-
.../paimon/flink/sink/WriterManagedMemoryTest.java | 237 +++++++++++++++++++++
8 files changed, 412 insertions(+), 6 deletions(-)
diff --git a/docs/docs/flink/configuration.md b/docs/docs/flink/configuration.md
index 4cddc0b102..4ece3c5256 100644
--- a/docs/docs/flink/configuration.md
+++ b/docs/docs/flink/configuration.md
@@ -61,6 +61,36 @@ INSERT INTO paimon_table /*+ OPTIONS(
SELECT * FROM source_table;
```
+### Managed Memory for an Explicit Writer Resource Group
+
+When setting `sink.writer-memory` to configure the writer slot sharing group's
task heap,
+use `sink.writer-managed-memory` to also request managed memory for that
group. This optional
+budget must be positive and requires `sink.writer-memory`; `sink.writer-cpu`
retains its default
+of one core unless explicitly configured.
+See Flink's [fine-grained resource
management](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/finegrained_resource/)
+for the deployment requirements and limitations of explicit slot sharing group
resources.
+
+```sql
+INSERT INTO paimon_table /*+ OPTIONS(
+ 'sink.writer-memory' = '512 mb',
+ 'sink.writer-managed-memory' = '1 gb',
+ 'sink.use-managed-memory-allocator' = 'true',
+ 'sink.managed.writer-buffer-memory' = '256 mb'
+) */
+SELECT * FROM source_table;
+```
+
+The managed budget is a resource request for each shared slot of the writer
group, not a total
+across all writer subtasks or a guaranteed allocation to the writer alone.
Actual allocation
+depends on Flink's resource management mode, available resources, and managed
memory consumers.
+The writer-buffer setting remains a weight: `256 mb` in this example does not
cap the writer's
+buffer at 256 MiB. Flink determines each writer subtask's share of the slot's
managed memory.
+
+Setting the budget does not enable the managed allocator. With the allocator
disabled, writer
+buffers still use heap memory. The existing downstream slot sharing and
explicit committer
+resource settings are unchanged. Leaving the new option unset preserves
existing resource
+configuration, including default-slot managed allocation.
+
## Setting dynamic options
Dynamic options tune a query or a subsequently submitted job without changing
the options
diff --git a/docs/generated/flink_connector_configuration.html
b/docs/generated/flink_connector_configuration.html
index 37638d7f99..b713195233 100644
--- a/docs/generated/flink_connector_configuration.html
+++ b/docs/generated/flink_connector_configuration.html
@@ -368,6 +368,12 @@ under the License.
<td>Double</td>
<td>Sink writer cpu to control cpu cores of writer.</td>
</tr>
+ <tr>
+ <td><h5>sink.writer-managed-memory</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>Managed memory resource budget for the writer slot sharing
group. Must be positive and requires sink.writer-memory. This does not enable
sink.use-managed-memory-allocator or change the allocation weight
sink.managed.writer-buffer-memory. The actual writer buffer size depends on the
slot's allocated resources and managed memory consumers.</td>
+ </tr>
<tr>
<td><h5>sink.writer-memory</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
index b560920a91..fcf9f2745a 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
@@ -72,6 +72,7 @@ public class FlinkCdcMultiTableSink implements Serializable {
private final CatalogLoader catalogLoader;
private final double writeCpuCores;
private final MemorySize writeHeapMemory;
+ @Nullable private final MemorySize writeManagedMemory;
private final double commitCpuCores;
@Nullable private final MemorySize commitHeapMemory;
private final String commitUser;
@@ -99,6 +100,7 @@ public class FlinkCdcMultiTableSink implements Serializable {
this.databaseName = null;
this.writeCpuCores = writeCpuCores;
this.writeHeapMemory = writeHeapMemory;
+ this.writeManagedMemory = null;
this.commitCpuCores = commitCpuCores;
this.commitHeapMemory = commitHeapMemory;
this.commitUser = commitUser;
@@ -116,10 +118,35 @@ public class FlinkCdcMultiTableSink implements
Serializable {
String commitUser,
boolean eagerInit,
TableFilter tableFilter) {
+ this(
+ catalogLoader,
+ databaseName,
+ writeCpuCores,
+ writeHeapMemory,
+ null,
+ commitCpuCores,
+ commitHeapMemory,
+ commitUser,
+ eagerInit,
+ tableFilter);
+ }
+
+ public FlinkCdcMultiTableSink(
+ CatalogLoader catalogLoader,
+ String databaseName,
+ double writeCpuCores,
+ @Nullable MemorySize writeHeapMemory,
+ @Nullable MemorySize writeManagedMemory,
+ double commitCpuCores,
+ @Nullable MemorySize commitHeapMemory,
+ String commitUser,
+ boolean eagerInit,
+ TableFilter tableFilter) {
this.catalogLoader = catalogLoader;
this.databaseName = Preconditions.checkNotNull(databaseName);
this.writeCpuCores = writeCpuCores;
this.writeHeapMemory = writeHeapMemory;
+ this.writeManagedMemory = writeManagedMemory;
this.commitCpuCores = commitCpuCores;
this.commitHeapMemory = commitHeapMemory;
this.commitUser = commitUser;
@@ -195,7 +222,7 @@ public class FlinkCdcMultiTableSink implements Serializable
{
shuffled.transform(
WRITER_NAME, typeInfo,
createWriteOperator(sinkProvider, commitUser));
forwardParallelism(written, shuffled);
- configureSlotSharingGroup(written, writeCpuCores, writeHeapMemory);
+ configureSlotSharingGroup(written, writeCpuCores, writeHeapMemory,
writeManagedMemory);
// shuffle committables by table
DataStream<MultiTableCommittable> partitioned =
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
index 30a6a98d47..2e44c88bbe 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -69,6 +69,7 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
@Nullable private Integer parallelism;
private double writerCpu;
@Nullable private MemorySize writerMemory;
+ @Nullable private MemorySize writerManagedMemory;
private double committerCpu;
@Nullable private MemorySize committerMemory;
@@ -111,6 +112,7 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
this.parallelism = options.get(FlinkConnectorOptions.SINK_PARALLELISM);
this.writerCpu = options.get(FlinkConnectorOptions.SINK_WRITER_CPU);
this.writerMemory =
options.get(FlinkConnectorOptions.SINK_WRITER_MEMORY);
+ this.writerManagedMemory =
options.get(FlinkConnectorOptions.SINK_WRITER_MANAGED_MEMORY);
this.committerCpu =
options.get(FlinkConnectorOptions.SINK_COMMITTER_CPU);
this.committerMemory =
options.get(FlinkConnectorOptions.SINK_COMMITTER_MEMORY);
this.commitUser = createCommitUser(options);
@@ -191,6 +193,7 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
database,
writerCpu,
writerMemory,
+ writerManagedMemory,
committerCpu,
committerMemory,
commitUser,
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
index 741a0fb9d1..cee9df8666 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
@@ -20,17 +20,24 @@ package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.FlinkConnectorOptions;
+import org.apache.paimon.flink.action.MultiTablesSinkMode;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
+import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
@@ -123,6 +130,65 @@ public class FlinkCdcMultiTableSinkTest {
.hasMessageContaining("database name");
}
+ @Test
+ public void testDatabaseBuilderPropagatesWriterManagedMemory(
+ @TempDir java.nio.file.Path tempDir) {
+ Options catalogOptions = new Options();
+ catalogOptions.setString("warehouse", tempDir.toString());
+ Options options = new Options();
+ options.setString("sink.writer-memory", "512 mb");
+ options.setString("sink.writer-managed-memory", "128 mb");
+ options.setString("sink.committer-memory", "64 mb");
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ new FlinkCdcSyncDatabaseSinkBuilder<TestCdcEvent>()
+ .withInput(
+ env.fromData(
+ TestCdcEvent.class,
+ new TestCdcEvent("t", Collections.emptyList(),
0)))
+ .withParserFactory(TestCdcEventParser::new)
+ .withTableOptions(options)
+ .withDatabase("test_db")
+ .withCatalogLoader(() ->
FlinkCatalogFactory.createPaimonCatalog(catalogOptions))
+ .withTypeMapping(TypeMapping.defaultMapping())
+ .withMode(MultiTablesSinkMode.COMBINED)
+ .build();
+ StreamGraph graph = env.getStreamGraph();
+ ResourceProfile writerResources =
+ graph.getSlotSharingGroupResource("CDC MultiplexWriter").get();
+
assertThat(writerResources.getManagedMemory().getMebiBytes()).isEqualTo(128);
+
assertThat(writerResources.getTaskHeapMemory().getMebiBytes()).isEqualTo(512);
+ ResourceProfile committerResources =
+ graph.getSlotSharingGroupResource("Multiplex Global
Committer").get();
+ assertThat(committerResources.getManagedMemory().getBytes()).isZero();
+
assertThat(committerResources.getTaskHeapMemory().getMebiBytes()).isEqualTo(64);
+ }
+
+ @Test
+ public void testManagedMemoryWithoutHeapIsRejected() {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ FlinkCdcMultiTableSink sink =
+ new FlinkCdcMultiTableSink(
+ () -> FlinkCatalogFactory.createPaimonCatalog(new
Options()),
+ "test_db",
+ 1.0,
+ null,
+ MemorySize.ofMebiBytes(128),
+ 1.0,
+ null,
+ "test-user",
+ false,
+ null);
+ assertThatThrownBy(
+ () ->
+ sink.sinkFrom(
+ env.fromData(
+ CdcMultiplexRecord.class,
+ new CdcMultiplexRecord("", "",
null))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("sink.writer-managed-memory")
+ .hasMessageContaining("sink.writer-memory");
+ }
+
@Test
public void testDatabaseAwareConstructorRejectsNullDatabase() {
assertThatThrownBy(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
index 7bb4f5c468..0deded6d31 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
@@ -364,6 +364,17 @@ public class FlinkConnectorOptions {
.noDefaultValue()
.withDescription("Sink writer memory to control heap
memory of writer.");
+ public static final ConfigOption<MemorySize> SINK_WRITER_MANAGED_MEMORY =
+ ConfigOptions.key("sink.writer-managed-memory")
+ .memoryType()
+ .noDefaultValue()
+ .withDescription(
+ "Managed memory resource budget for the writer
slot sharing group. "
+ + "Must be positive and requires
sink.writer-memory. This does not "
+ + "enable
sink.use-managed-memory-allocator or change the allocation "
+ + "weight
sink.managed.writer-buffer-memory. The actual writer buffer "
+ + "size depends on the slot's allocated
resources and managed memory consumers.");
+
public static final ConfigOption<Double> SINK_COMMITTER_CPU =
ConfigOptions.key("sink.committer-cpu")
.doubleType()
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index c28d280128..21e9b722c4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -65,6 +65,7 @@ import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_MANAGED_WRITER_
import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_OPERATOR_UID_SUFFIX;
import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_USE_MANAGED_MEMORY;
import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_WRITER_CPU;
+import static
org.apache.paimon.flink.FlinkConnectorOptions.SINK_WRITER_MANAGED_MEMORY;
import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_WRITER_MEMORY;
import static org.apache.paimon.flink.FlinkConnectorOptions.generateCustomUid;
import static
org.apache.paimon.flink.utils.ManagedMemoryUtils.declareManagedMemory;
@@ -167,7 +168,10 @@ public abstract class FlinkSink<T> implements Serializable
{
}
configureSlotSharingGroup(
- written, options.get(SINK_WRITER_CPU),
options.get(SINK_WRITER_MEMORY));
+ written,
+ options.get(SINK_WRITER_CPU),
+ options.get(SINK_WRITER_MEMORY),
+ options.get(SINK_WRITER_MANAGED_MEMORY));
if (!table.primaryKeys().isEmpty() && options.get(PRECOMMIT_COMPACT)) {
SingleOutputStreamOperator<Committable> beforeSort =
@@ -295,18 +299,40 @@ public abstract class FlinkSink<T> implements
Serializable {
SingleOutputStreamOperator<?> operator,
double cpuCores,
@Nullable MemorySize heapMemory) {
+ configureSlotSharingGroup(operator, cpuCores, heapMemory, null);
+ }
+
+ public static void configureSlotSharingGroup(
+ SingleOutputStreamOperator<?> operator,
+ double cpuCores,
+ @Nullable MemorySize heapMemory,
+ @Nullable MemorySize managedMemory) {
+ if (managedMemory != null) {
+ checkArgument(
+ heapMemory != null,
+ "'%s' requires '%s' to configure the writer slot sharing
group.",
+ SINK_WRITER_MANAGED_MEMORY.key(),
+ SINK_WRITER_MEMORY.key());
+ checkArgument(
+ managedMemory.getBytes() > 0,
+ "'%s' must be positive.",
+ SINK_WRITER_MANAGED_MEMORY.key());
+ }
if (heapMemory == null) {
return;
}
- SlotSharingGroup slotSharingGroup =
+ SlotSharingGroup.Builder builder =
SlotSharingGroup.newBuilder(operator.getName())
.setCpuCores(cpuCores)
.setTaskHeapMemory(
new org.apache.flink.configuration.MemorySize(
- heapMemory.getBytes()))
- .build();
- operator.slotSharingGroup(slotSharingGroup);
+ heapMemory.getBytes()));
+ if (managedMemory != null) {
+ builder.setManagedMemory(
+ new
org.apache.flink.configuration.MemorySize(managedMemory.getBytes()));
+ }
+ operator.slotSharingGroup(builder.build());
}
public static void assertStreamingConfiguration(StreamExecutionEnvironment
env) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/WriterManagedMemoryTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/WriterManagedMemoryTest.java
new file mode 100644
index 0000000000..fe7e106e80
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/WriterManagedMemoryTest.java
@@ -0,0 +1,237 @@
+/*
+ * 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.paimon.flink.sink;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the writer slot sharing group's managed memory budget. */
+class WriterManagedMemoryTest extends CommitterTestBase {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testManagedResourcesAndDownstreamInheritance(boolean
managedAllocator) throws Exception {
+ FileStoreTable table = createTable("128 mb", managedAllocator);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream<Committable> written = write(env, table);
+ StreamGraph graph = env.getStreamGraph();
+ String writerGroup = written.getTransformation().getName();
+ ResourceProfile resources =
graph.getSlotSharingGroupResource(writerGroup).get();
+
assertThat(resources.getManagedMemory()).isEqualTo(MemorySize.ofMebiBytes(128));
+
assertThat(resources.getTaskHeapMemory()).isEqualTo(MemorySize.ofMebiBytes(512));
+
assertThat(resources.getCpuCores().getValue().doubleValue()).isEqualTo(2.5);
+
assertThat(resources.getTaskOffHeapMemory()).isEqualTo(MemorySize.ZERO);
+
assertThat(written.getTransformation().getManagedMemoryOperatorScopeUseCaseWeights())
+ .isEqualTo(
+ managedAllocator
+ ?
Collections.singletonMap(ManagedMemoryUseCase.OPERATOR, 256)
+ : Collections.emptyMap());
+ graph.getStreamNodes().stream()
+ .filter(n -> !n.getOperatorName().startsWith("Source:"))
+ .forEach(n ->
assertThat(n.getSlotSharingGroup()).isEqualTo(writerGroup));
+
assertThat(writerVertex(graph).getSlotSharingGroup().getResourceProfile())
+ .isEqualTo(resources);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testAbsentOptionPreservesExistingResources(boolean explicitHeap)
throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(
+ options -> {
+
options.setString("sink.use-managed-memory-allocator", "true");
+ if (explicitHeap) {
+ options.setString("sink.writer-memory", "512
mb");
+ }
+ });
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ write(env, table);
+ ResourceProfile resources =
+
writerVertex(env.getStreamGraph()).getSlotSharingGroup().getResourceProfile();
+ if (explicitHeap) {
+
assertThat(resources.getManagedMemory()).isEqualTo(MemorySize.ZERO);
+ } else {
+ assertThat(resources).isEqualTo(ResourceProfile.UNKNOWN);
+ }
+ }
+
+ @Test
+ void testManagedMemoryRequiresHeap() throws Exception {
+ FileStoreTable table =
+ createFileStoreTable(
+ options ->
options.setString("sink.writer-managed-memory", "128 mb"));
+ assertThatThrownBy(() ->
write(StreamExecutionEnvironment.getExecutionEnvironment(), table))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("sink.writer-managed-memory")
+ .hasMessageContaining("sink.writer-memory");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testManagedMemoryMustBePositive(boolean managedAllocator) throws
Exception {
+ FileStoreTable table = createTable("0 b", managedAllocator);
+ assertThatThrownBy(() ->
write(StreamExecutionEnvironment.getExecutionEnvironment(), table))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("sink.writer-managed-memory")
+ .hasMessageContaining("positive");
+ }
+
+ @Test
+ void testExplicitCommitterResourcesArePreserved() throws Exception {
+ Options options = new Options();
+ options.setString("sink.committer-memory", "64 mb");
+ options.setString("sink.committer-cpu", "0.25");
+ FileStoreTable table = createTable("128 mb",
true).copy(options.toMap());
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ write(env, table);
+ StreamGraph graph = env.getStreamGraph();
+ String group =
+ graph.getStreamNodes().stream()
+ .filter(n -> n.getOperatorName().startsWith("Global
Committer"))
+ .findFirst()
+ .get()
+ .getSlotSharingGroup();
+ ResourceProfile resources =
graph.getSlotSharingGroupResource(group).get();
+
assertThat(resources.getTaskHeapMemory()).isEqualTo(MemorySize.ofMebiBytes(64));
+ assertThat(resources.getManagedMemory()).isEqualTo(MemorySize.ZERO);
+
assertThat(resources.getCpuCores().getValue().doubleValue()).isEqualTo(0.25);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testWriterUsesManagedBudgetFromJobGraph(boolean primaryKey) throws
Exception {
+ FileStoreTable table = createTable("4 mb", true, primaryKey);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream<Committable> written = write(env, table);
+ JobVertex vertex = writerVertex(env.getStreamGraph());
+ long managedBytes =
+
vertex.getSlotSharingGroup().getResourceProfile().getManagedMemory().getBytes();
+ assertThat(managedBytes).isEqualTo(4L * 1024 * 1024);
+ StreamConfig config = new StreamConfig(vertex.getConfiguration());
+ double fraction =
+ config.getManagedMemoryFractionOperatorUseCaseOfSlot(
+ ManagedMemoryUseCase.OPERATOR,
+ new Configuration(),
+ new Configuration(),
+ getClass().getClassLoader());
+ assertThat(fraction).isEqualTo(1.0);
+ OneInputStreamOperatorFactory<InternalRow, Committable> factory =
+ (OneInputStreamOperatorFactory<InternalRow, Committable>)
+ ((OneInputTransformation<InternalRow, Committable>)
+ written.getTransformation())
+ .getOperatorFactory();
+ try (MockEnvironment environment =
+
MockEnvironment.builder().setManagedMemorySize(managedBytes).build();
+ OneInputStreamOperatorTestHarness<InternalRow, Committable>
harness =
+ new OneInputStreamOperatorTestHarness<>(factory,
environment)) {
+ // The harness declares STATE_BACKEND by default; the generated
writer group does not.
+ harness.getStreamConfig().setStateBackendUsesManagedMemory(false);
+ harness.getStreamConfig()
+ .setManagedMemoryFractionOperatorOfUseCase(
+ ManagedMemoryUseCase.OPERATOR, fraction);
+ harness.setup(new CommittableTypeInfo().createSerializer(new
ExecutionConfig()));
+ harness.open();
+ harness.processElement(GenericRow.of(1, 1L), 1);
+ PrepareCommitOperator<?, ?> operator =
+ (PrepareCommitOperator<?, ?>)
harness.getOneInputOperator();
+
assertThat(operator.memoryPoolFactory.totalBufferSize()).isEqualTo(managedBytes);
+
assertThat(operator.memoryPoolFactory.usedBufferSize()).isPositive();
+ harness.prepareSnapshotPreBarrier(1);
+ assertThat(harness.extractOutputValues()).isNotEmpty();
+ }
+ }
+
+ private FileStoreTable createTable(String managedMemory, boolean
managedAllocator)
+ throws Exception {
+ return createTable(managedMemory, managedAllocator, false);
+ }
+
+ private FileStoreTable createTable(
+ String managedMemory, boolean managedAllocator, boolean
primaryKey) throws Exception {
+ Options options = new Options();
+ options.setString("path", tablePath.toString());
+ options.setString("bucket", "1");
+ options.setString("bucket-key", "a");
+ options.setString("sink.writer-cpu", "2.5");
+ options.setString("sink.writer-memory", "512 mb");
+ options.setString("sink.writer-managed-memory", managedMemory);
+ options.setString("sink.use-managed-memory-allocator",
String.valueOf(managedAllocator));
+ options.setString("write-buffer-for-append", "true");
+ Schema.Builder schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.BIGINT())
+ .options(options.toMap());
+ if (primaryKey) {
+ schema.primaryKey("a");
+ }
+ new FileSystemSchemaManager(LocalFileIO.create(),
tablePath).createTable(schema.build());
+ return FileStoreTableFactory.create(LocalFileIO.create(), options);
+ }
+
+ private DataStream<Committable> write(StreamExecutionEnvironment env,
FileStoreTable table) {
+ DataStream<InternalRow> source =
+ env.fromData(InternalRow.class, GenericRow.of(1, 1L))
+ .name("upstream")
+ .slotSharingGroup("main");
+ FlinkSink<InternalRow> sink = new FixedBucketSink(table, null);
+ DataStream<Committable> written = sink.doWrite(source, "test-user", 2);
+ sink.doCommit(written, "test-user");
+ return written;
+ }
+
+ private JobVertex writerVertex(StreamGraph graph) {
+ for (JobVertex vertex : graph.getJobGraph().getVertices()) {
+ if (vertex.getName().startsWith("Writer")) {
+ return vertex;
+ }
+ }
+ throw new AssertionError("Writer vertex not found");
+ }
+}
