hudi-agent commented on code in PR #19920:
URL: https://github.com/apache/hudi/pull/19920#discussion_r3995723051
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java:
##########
@@ -721,11 +724,29 @@ private static DataStream<HoodieFlinkInternalRow>
createBucketAssignStream(
* @return the compaction pipeline
*/
public static DataStreamSink<CompactionCommitEvent> compact(Configuration
conf, DataStream<RowData> dataStream) {
- DataStreamSink<CompactionCommitEvent> compactionCommitEventDataStream =
dataStream.transform("compact_plan_generate",
+ SingleOutputStreamOperator<CompactionPlanEvent> compactionPlanStream =
dataStream.transform("compact_plan_generate",
TypeInformation.of(CompactionPlanEvent.class),
new CompactionPlanOperator(conf))
.setParallelism(1) // plan generate must be singleton
- .setMaxParallelism(1)
+ .setMaxParallelism(1);
+ if
(conf.getOptional(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP).isPresent())
{
+ String slotSharingGroup =
conf.get(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP);
+ SlotSharingGroup.Builder slotSharingGroupBuilder =
SlotSharingGroup.newBuilder(slotSharingGroup);
+ // Flink requires cpu cores and task heap memory to be configured
together when specifying
+ // resources for a slot sharing group, reuse the TaskManager-level
configuration for both.
+ Option<Double> cpuCores =
Option.fromJavaOptional(conf.getOptional(TaskManagerOptions.CPU_CORES));
+ Option<MemorySize> taskHeapMemory =
Option.fromJavaOptional(conf.getOptional(TaskManagerOptions.TASK_HEAP_MEMORY));
+ if (!taskHeapMemory.isPresent()) {
+ taskHeapMemory = deriveTaskHeapMemory(conf);
+ }
+ if (cpuCores.isPresent() && taskHeapMemory.isPresent()) {
+
slotSharingGroupBuilder.setCpuCores(cpuCores.get()).setTaskHeapMemory(taskHeapMemory.get());
Review Comment:
🤖 Reusing the TaskManager totals (`taskmanager.cpu.cores`, full task heap)
as the slot group's resource spec means this parallelism-1 operator requests an
entire TM's CPU and task heap for its slot. The RM only fulfills a request when
a TM's total/available profile is `allFieldsNoLessThan` it, so on a
standalone/session cluster with `numberOfTaskSlots > 1` (or any TM already
hosting default-group slots) this can never be matched and the job fails with
`NoResourceAvailableException`; in fine-grained mode it monopolizes a whole TM.
Would a fraction (or dedicated
`compaction.plan_generate.slot_sharing_group.{cpu,memory}` options) be a safer
definition of the slot's resources?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java:
##########
@@ -721,11 +724,29 @@ private static DataStream<HoodieFlinkInternalRow>
createBucketAssignStream(
* @return the compaction pipeline
*/
public static DataStreamSink<CompactionCommitEvent> compact(Configuration
conf, DataStream<RowData> dataStream) {
- DataStreamSink<CompactionCommitEvent> compactionCommitEventDataStream =
dataStream.transform("compact_plan_generate",
+ SingleOutputStreamOperator<CompactionPlanEvent> compactionPlanStream =
dataStream.transform("compact_plan_generate",
TypeInformation.of(CompactionPlanEvent.class),
new CompactionPlanOperator(conf))
.setParallelism(1) // plan generate must be singleton
- .setMaxParallelism(1)
+ .setMaxParallelism(1);
+ if
(conf.getOptional(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP).isPresent())
{
+ String slotSharingGroup =
conf.get(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP);
Review Comment:
🤖 nit: this inline block (resolving cpu/heap, building and registering the
slot sharing group) adds a good chunk of unrelated logic into `compact()`.
Might be worth pulling it into a small helper like
`configureSlotSharingGroup(conf, compactionPlanStream)` so `compact()` stays
focused on wiring the pipeline stages together.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java:
##########
@@ -721,11 +725,29 @@ private static DataStream<HoodieFlinkInternalRow>
createBucketAssignStream(
* @return the compaction pipeline
*/
public static DataStreamSink<CompactionCommitEvent> compact(Configuration
conf, DataStream<RowData> dataStream) {
- DataStreamSink<CompactionCommitEvent> compactionCommitEventDataStream =
dataStream.transform("compact_plan_generate",
+ SingleOutputStreamOperator<CompactionPlanEvent> compactionPlanStream =
dataStream.transform("compact_plan_generate",
TypeInformation.of(CompactionPlanEvent.class),
new CompactionPlanOperator(conf))
.setParallelism(1) // plan generate must be singleton
- .setMaxParallelism(1)
+ .setMaxParallelism(1);
+ if
(conf.getOptional(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP).isPresent())
{
+ String slotSharingGroup =
conf.get(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP);
+ SlotSharingGroup.Builder slotSharingGroupBuilder =
SlotSharingGroup.newBuilder(slotSharingGroup);
+ // Flink requires cpu cores and task heap memory to be configured
together when specifying
+ // resources for a slot sharing group, reuse the TaskManager-level
configuration for both.
+ Optional<Double> cpuCores =
conf.getOptional(TaskManagerOptions.CPU_CORES);
+ Optional<MemorySize> taskHeapMemory =
conf.getOptional(TaskManagerOptions.TASK_HEAP_MEMORY);
+ if (!taskHeapMemory.isPresent()) {
+ taskHeapMemory = deriveTaskHeapMemory(conf);
+ }
+ if (cpuCores.isPresent() && taskHeapMemory.isPresent()) {
+
slotSharingGroupBuilder.setCpuCores(cpuCores.get()).setTaskHeapMemory(taskHeapMemory.get());
+ }
+ // the slot sharing group must be registered with the environment before
it can be referenced by name
+
dataStream.getExecutionEnvironment().registerSlotSharingGroup(slotSharingGroupBuilder.build());
+ compactionPlanStream =
compactionPlanStream.slotSharingGroup(slotSharingGroup);
Review Comment:
🤖 Confirmed against Flink 1.20.1: `StreamGraph.getSlotSharingGroup(id)`
resolves the `partitionCustom` virtual node back to `compact_plan_generate`, so
`compact_task` and `compact_commit` both inherit the new group. Since the group
also carries a resource profile of the full TM cpu + task heap, that means each
of the `compaction.tasks` subtasks would demand an exclusive slot with that
profile — on a single-TM setup the job likely can't be scheduled at all.
Pinning the two downstream operators to `"default"` (or the write pipeline's
group) should address both the isolation and the scheduling side of it.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java:
##########
@@ -721,11 +725,29 @@ private static DataStream<HoodieFlinkInternalRow>
createBucketAssignStream(
* @return the compaction pipeline
*/
public static DataStreamSink<CompactionCommitEvent> compact(Configuration
conf, DataStream<RowData> dataStream) {
- DataStreamSink<CompactionCommitEvent> compactionCommitEventDataStream =
dataStream.transform("compact_plan_generate",
+ SingleOutputStreamOperator<CompactionPlanEvent> compactionPlanStream =
dataStream.transform("compact_plan_generate",
TypeInformation.of(CompactionPlanEvent.class),
new CompactionPlanOperator(conf))
.setParallelism(1) // plan generate must be singleton
- .setMaxParallelism(1)
+ .setMaxParallelism(1);
+ if
(conf.getOptional(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP).isPresent())
{
+ String slotSharingGroup =
conf.get(FlinkOptions.COMPACTION_PLAN_GENERATE_SLOT_SHARING_GROUP);
+ SlotSharingGroup.Builder slotSharingGroupBuilder =
SlotSharingGroup.newBuilder(slotSharingGroup);
+ // Flink requires cpu cores and task heap memory to be configured
together when specifying
+ // resources for a slot sharing group, reuse the TaskManager-level
configuration for both.
+ Optional<Double> cpuCores =
conf.getOptional(TaskManagerOptions.CPU_CORES);
+ Optional<MemorySize> taskHeapMemory =
conf.getOptional(TaskManagerOptions.TASK_HEAP_MEMORY);
Review Comment:
🤖 There's an existing precedent for this in
`HoodieTableSink#getSinkRuntimeProvider`, which reads cluster-level settings
via `dataStream.getExecutionEnvironment().getConfiguration()`
(`OptionsInference.setupRuntimeConfigs`). `StreamContextEnvironment` merges the
cluster flink-conf into that configuration, so falling back to it for
`TaskManagerOptions.*` would pick up `taskmanager.cpu.cores` / memory settings
without users having to copy them into the table options.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
