hudi-agent commented on code in PR #19023: URL: https://github.com/apache/hudi/pull/19023#discussion_r3436928750
########## hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/TxnStateMemo.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.util; + +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.Pair; + +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Memorizes transaction states by instant for Flink streaming writes. + */ +public class TxnStateMemo { + + private final Map<String, TxnState> memo = new HashMap<>(); + + public void memo(String instant, + Option<Pair<HoodieInstant, Map<String, String>>> lastCompletedTxnAndMetadata, + Set<String> conflictResolutionExclusionInstants, + Set<String> pendingInflightAndRequestedInstants) { + memo.put(instant, new TxnState( + lastCompletedTxnAndMetadata, + new HashSet<>(conflictResolutionExclusionInstants), + new HashSet<>(pendingInflightAndRequestedInstants))); + } + + public Option<TxnState> get(String instant) { + return Option.ofNullable(memo.get(instant)); + } + + public void slip(String instant) { Review Comment: 🤖 nit: `slip` doesn't carry an obvious meaning here — `remove` or `evict` would tell a reader immediately what this does without needing to look at the body. <sub><i>- AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/TxnStateMemo.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.util; + +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.Pair; + +import lombok.AccessLevel; +import lombok.AllArgsConstructor; +import lombok.Getter; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Memorizes transaction states by instant for Flink streaming writes. + */ +public class TxnStateMemo { + + private final Map<String, TxnState> memo = new HashMap<>(); + + public void memo(String instant, Review Comment: 🤖 nit: could you rename this to `record` or `put`? Using a noun as a verb (`txnStateMemo.memo(...)` at the call site) is unusual in Java APIs and reads a bit awkwardly. <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/StreamWriteOperatorCoordinator.java: ########## @@ -547,10 +546,14 @@ private boolean recommitInstant(HoodieTimeline completedTimeline, long checkpoin if (writeClient.getConfig().getFailedWritesCleanPolicy().isLazy()) { writeClient.getHeartbeatClient().start(instant); } + // Initialize the transaction state so same-writer instants can be excluded + // during OCC conflict resolution. + writeClient.preTxn(tableState.operationType, this.metaClient, instant, this.eventBuffers.getAllInstants()); Review Comment: 🤖 I think the multi-pending recommit case is still broken — could you double-check? With A and B both inflight on restart, after A's recommit succeeds, `doCommit` calls `eventBuffers.reset(ckp_A)`, so when B's `preTxn` runs next, `eventBuffers.getAllInstants()` returns `{B}` only — A is no longer in the exclusion set. The coordinator's `metaClient` isn't reloaded between iterations either (the single `reloadActiveTimeline()` happens once at restoreEvents entry), so `getLastCompletedTxnInstantAndMetadata(metaClient)` still returns the pre-A txn as the cutoff. Inside the lock, `preCommit` builds a fresh table whose timeline now shows A completed; `getCandidateInstants` returns A via `findInstantsAfter(preA.requestedTime())`, the exclusion filter drops only B, and A vs B (same file groups from the same writer) throws `HoodieWriteConflictException` — exactly the failure this PR is fixing. Snapshotting `eventBuffers.getAllInstants()` once before the loop (or calling `metaClien t.reloadActiveTimeline()` before each `preTxn`) would close this. @danny0405 could you sanity-check this trace? <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]
