hudi-agent commented on code in PR #19952:
URL: https://github.com/apache/hudi/pull/19952#discussion_r4012467193
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java:
##########
@@ -260,11 +271,93 @@ public void testReceiveInvalidEvent() {
"Receive an unexpected event for instant abc from task 0");
}
+ @Test
+ void testDeferredRecommitAfterScaleUp() throws Exception {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ String restoredInstant = restoreFirstCheckpointAfterScaleUp(conf);
+ String nextInstant = requestInstantTime(1);
+ assertNotEquals(restoredInstant, nextInstant);
+ assertEquals(nextInstant, coordinator.getInstant());
+ coordinator.checkpointCoordinator(2, new CompletableFuture<>());
+ sendCheckpointEvents(1, nextInstant, 4);
+
+ coordinator.notifyCheckpointComplete(2);
+
+ HoodieTimeline completed = StreamerUtil.createMetaClient(conf)
+ .reloadActiveTimeline().filterCompletedInstants();
+ assertTrue(completed.containsInstant(restoredInstant));
+ assertTrue(completed.containsInstant(nextInstant));
+ assertEquals(2,
completed.readCommitMetadata(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
restoredInstant))
+ .getPartitionToWriteStats().size(), "Both original writers must be
included in the restored commit");
+ assertEquals(4,
completed.readCommitMetadata(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
nextInstant))
+ .getPartitionToWriteStats().size(), "All scaled-up writers must be
included in the next commit");
+ assertNull(coordinator.getEventBuffer(-1));
+ assertNull(coordinator.getEventBuffer(1));
+ assertNull(((MockOperatorCoordinatorContext)
coordinator.getContext()).getJobFailureReason());
+ }
+
+ @Disabled("https://github.com/apache/hudi/issues/19922: deferred bootstrap
metadata is omitted from the next checkpoint")
Review Comment:
🤖 Since this PR turns the previous hard failure into a path that proceeds
via deferred commit, the gap this disabled test documents becomes a
silent-data-loss window (instant eventually rolled back while source offsets
have advanced) rather than a crash. Is there a tracking issue for it, and would
it be reasonable to persist bootstrap-populated buffers in
`getAllCompletedEvents` as an interim so a second restart in the window doesn't
drop the batch?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.hudi.sink.common;
+
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.sink.event.Correspondent;
+import org.apache.hudi.sink.event.WriteMetadataEvent;
+import org.apache.hudi.sink.utils.MockOperatorStateStore;
+import org.apache.hudi.util.FlinkWriteClients;
+import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.RuntimeContextUtils;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import
org.apache.flink.streaming.api.operators.collect.utils.MockFunctionSnapshotContext;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedStatic;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.OptionalLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests checkpoint identity and bootstrap events independently of the writer
implementation.
+ */
+class TestAbstractStreamWriteFunction {
+ private final Configuration conf = new Configuration();
+ private final JobID jobId = new JobID();
+ private final List<OperatorEvent> events = new ArrayList<>();
+
+ private final MockOperatorStateStore stateStore = new
MockOperatorStateStore();
+ private final HoodieTimeline pendingTimeline = mock(HoodieTimeline.class);
+ private final Correspondent correspondent = mock(Correspondent.class);
+ private final TestWriteFunction function = new TestWriteFunction(conf);
+
+ @AfterEach
+ void tearDown() throws Exception {
+ function.close();
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, 1})
+ void testFreshStartUsesInitialCheckpointId(int attempt) throws Exception {
+ initialize(-1L, attempt);
+
+ assertEquals("002", function.instantToWrite(true));
+ verify(correspondent).requestInstantTime(-1L);
+ if (attempt == 0) {
+ assertTrue(events.isEmpty());
+ } else {
+ assertCleanupEvent(-1L);
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"0, same", "0, different", "0, missing", "1, same", "1,
different", "1, missing"})
+ void testRestoredCheckpointId(int attempt, String savedJob) throws Exception
{
Review Comment:
🤖 nit: `savedJob` being a stringly-typed "same"/"different"/"missing"
parameter is a bit hard to follow at the call site (`@CsvSource({"0, same", "0,
different", ...})`). Consider a small enum (e.g.
`SavedJobState.SAME/DIFFERENT/MISSING`) so the intent is clearer without
reading into the method body.
<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/common/AbstractStreamWriteFunction.java:
##########
@@ -166,15 +160,13 @@ public void initializeState(FunctionInitializationContext
context) throws Except
"write-metadata-state",
TypeInformation.of(WriteMetadataEvent.class)
));
- this.jobIdState = context.getOperatorStateStore().getListState(
- new ListStateDescriptor<>(
- "job-id-state",
- TypeInformation.of(JobID.class)
- ));
int attemptId = RuntimeContextUtils.getAttemptNumber(getRuntimeContext());
if (context.isRestored()) {
- initCheckpointId(attemptId,
context.getRestoredCheckpointId().orElse(-1L));
+ // sets up the known checkpoint id as the last successful checkpoint id
for purposes of:
+ // 1). old events cleaning;
+ // 2). instant time request for current checkpoint.
+ this.checkpointId = context.getRestoredCheckpointId().orElse(-1L);
Review Comment:
🤖 Does the scale-up fix also hold for CoW upsert, where
`isBlockingInstantGeneration` is on? In that case the first restored writer's
request for key N hits `awaitAllInstantsToCompleteIfNecessary` → `blockFor` on
the partially-filled `-1` buffer, which can only be reset by the deferred
commit at `notifyCheckpointComplete(N+1)` — but that checkpoint can't complete
while the writer is blocked inside `instantToWrite`. It looks like it would
spin on `write.commit.ack.timeout` (5 min) and restart. The new scale-up tests
all go through `resetToMergeOnRead`, so this path isn't covered.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
