danny0405 commented on code in PR #18350:
URL: https://github.com/apache/hudi/pull/18350#discussion_r3766272833
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/TransactionManager.java:
##########
@@ -41,40 +46,166 @@ public class TransactionManager implements Serializable,
AutoCloseable {
protected final LockManager lockManager;
@Getter
protected final boolean isLockRequired;
+ private final transient TimeGenerator timeGenerator;
+ private volatile long lockHolderId; // lock holder ID
+ private int permits; // allows for nested transaction
protected Option<HoodieInstant> changeActionInstant = Option.empty();
private Option<HoodieInstant> lastCompletedActionInstant = Option.empty();
public TransactionManager(HoodieWriteConfig config, HoodieStorage storage) {
- this(new LockManager(config, storage), config.isLockRequired());
+ this(config, new LockManager(config, storage));
}
- protected TransactionManager(LockManager lockManager, boolean
isLockRequired) {
+ protected TransactionManager(HoodieWriteConfig writeConfig, LockManager
lockManager) {
+ this(lockManager, writeConfig.isLockRequired(),
TimeGenerators.getTimeGenerator(writeConfig.getTimeGeneratorConfig()));
+ }
+
+ public TransactionManager(LockManager lockManager, boolean isLockRequired,
TimeGenerator timeGenerator) {
this.lockManager = lockManager;
this.isLockRequired = isLockRequired;
+ this.timeGenerator = timeGenerator;
+ this.lockHolderId = -1;
+ this.permits = 0;
+ }
+
+ /**
+ * Caution: the invoker needs to ensure that API called within a lock
context.
+ */
+ public String generateInstantTime() {
+ if (lockHolderId < 0 && isLockRequired) {
+ throw new HoodieLockException("Cannot create instant without acquiring a
lock first.");
+ }
+ return HoodieInstantTimeGenerator.createNewInstantTime(timeGenerator, 0L);
+ }
+
+ /**
+ * Generates an instant time and executes an action that requires that
instant time within a lock.
+ * @param instantTimeConsumingAction a function that takes the generated
instant time and performs some action
+ * @return the result of the action
+ * @param <T> type of the result
+ */
+ public <T> T executeStateChangeWithInstant(Function<String, T>
instantTimeConsumingAction) {
+ return executeStateChangeWithInstant(Option.empty(), Option.empty(),
instantTimeConsumingAction);
+ }
+
+ /**
+ * Uses the provided instant if present or else generates an instant time
and executes an action that requires that instant time within a lock.
+ * @param providedInstantTime an optional instant time provided by the
caller. If not provided, a new instant time will be generated.
+ * @param instantTimeConsumingAction a function that takes the generated
instant time and performs some action
+ * @return the result of the action
+ * @param <T> type of the result
+ */
+ public <T> T executeStateChangeWithInstant(Option<String>
providedInstantTime, Function<String, T> instantTimeConsumingAction) {
+ return executeStateChangeWithInstant(providedInstantTime, Option.empty(),
instantTimeConsumingAction);
+ }
+
+ /**
+ * Uses the provided instant if present or else generates an instant time
and executes an action that requires that instant time within a lock.
+ * @param providedInstantTime an optional instant time provided by the
caller. If not provided, a new instant time will be generated.
+ * @param lastCompletedActionInstant optional input representing the last
completed instant, used for logging purposes.
+ * @param instantTimeConsumingAction a function that takes the generated
instant time and performs some action
+ * @return the result of the action
+ * @param <T> type of the result
+ */
+ public <T> T executeStateChangeWithInstant(Option<String>
providedInstantTime, Option<HoodieInstant> lastCompletedActionInstant,
Function<String, T> instantTimeConsumingAction) {
+ if (isLockRequired()) {
+ acquireLock();
+ }
+ String requestedInstant = providedInstantTime.orElseGet(() ->
HoodieInstantTimeGenerator.createNewInstantTime(timeGenerator, 0L));
+ try {
+ if (lastCompletedActionInstant.isEmpty()) {
+ LOG.info("State change starting for {}", changeActionInstant);
+ } else {
+ LOG.info("State change starting for {} with latest completed action
instant {}", changeActionInstant, lastCompletedActionInstant.get());
+ }
+ return instantTimeConsumingAction.apply(requestedInstant);
+ } finally {
+ if (isLockRequired()) {
+ releaseLock();
+ LOG.info("State change ended for {}", requestedInstant);
+ }
+ }
+ }
+
+ public void beginStateChange() {
+ beginStateChange(Option.empty(), Option.empty());
}
public void beginStateChange(Option<HoodieInstant> changeActionInstant,
Option<HoodieInstant>
lastCompletedActionInstant) {
if (isLockRequired) {
LOG.info("State change starting for {} with latest completed action
instant {}",
changeActionInstant, lastCompletedActionInstant);
- lockManager.lock();
+ acquireLock();
reset(this.changeActionInstant, changeActionInstant,
lastCompletedActionInstant);
LOG.info("State change started for {} with latest completed action
instant {}",
changeActionInstant, lastCompletedActionInstant);
}
}
+ public void endStateChange() {
+ endStateChange(Option.empty());
+ }
+
public void endStateChange(Option<HoodieInstant> changeActionInstant) {
if (isLockRequired) {
LOG.info("State change ending for action instant {}",
changeActionInstant);
if (reset(changeActionInstant, Option.empty(), Option.empty())) {
- lockManager.unlock();
+ releaseLock();
LOG.info("State change ended for action instant {}",
changeActionInstant);
}
}
}
+ /**
+ * Caution: the {@code hasLock} flag can not be used to skip the `#lock`
eagerly if the thread switches,
+ * the corner case below can cause deadlock:
+ *
+ * <pre>
+ * threadA => acquireLock(), {@code hasLock} setup as true and got the
lock acquired;
+ * threadB => acquireLock(), check the {@code hasLock} as true, returns
early;
+ * threadB => releaseLock(), set up the {@code hasLock} as false;
+ * threadA => releaseLock(), detect the {@code hasLock} as false and
returns early.
+ *
+ * The lock held by threadA will never be released.
+ * </pre>
+ */
+ private void acquireLock() {
+ if (lockHolderId > 0 && isLockHeldByCurrentThread()) {
+ LOG.info("{}: Lock already acquired, skipping lock acquisition.", this);
+ permits++;
+ return;
+ }
+ lockManager.lock();
+ permits++;
Review Comment:
have remove the reentrancy since it is hard to keep robust for correctness:
Short answer: I would make `TransactionManager` non-reentrant. `lockHolderId
+ permits` implements reentrant mutex mechanics, but it does not provide
coherent nested-transaction semantics.
The current implementation has a concrete lock leak:
1. `begin(A)` → `permits = 1`, owner instant A
2. `begin(B)` → `permits = 2`, owner instant overwritten with B
3. `end(A)` → `reset()` rejects A, so no permit is released
4. `end(B)` → releases one permit, leaving `permits = 1`
The existing [reentrancy
test](/Users/chenyuzhao/workspace/hudi-dev/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestTransactionManager.java:103)
uses exactly this sequence but only checks that calls do not throw. It never
verifies that the lock was released.
Other concerns:
- Nested `begin(B)` overwrites A’s transaction metadata; ending B does not
restore A.
- Clearing `lockHolderId` before calling the underlying `unlock()` leaves
local state incorrect if the provider fails to unlock.
- `Thread.getId()` may eventually be reused; storing the owning `Thread`
would be safer.
- `TransactionManager` is serializable, but `lockHolderId` and `permits` are
not transient, while `timeGenerator` is transient. Deserialization can
therefore produce meaningless ownership state and a null generator.
- `permits <= 0` hides under-release; it should enforce exact invariants.
I found one real nested use in
[BaseHoodieTableServiceClient.java](/Users/chenyuzhao/workspace/hudi-dev/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:1344):
an outer transaction calls `executeStateChangeWithInstant()` merely to
generate an instant. That can use `txnManager.generateInstantTime()` directly,
so it does not justify general reentrancy.
My recommendation:
- Make `beginStateChange`/`executeStateChangeWithInstant` fail fast when the
same thread already owns a transaction.
- Refactor helpers called inside transactions to reuse the existing
transaction explicitly.
- Prefer the scoped callback API over manual `begin`/`end`.
- Treat `begin(A) → begin(B)` as a programming error, especially when A and
B differ.
If true nested transactions are required, a counter is insufficient. You
would need a stack of transaction frames, strict LIFO completion, restoration
of outer metadata, and an actual locally synchronized ownership mechanism
around the distributed lock. That is considerably more complexity than the
apparent use cases warrant.
--
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]
