nsivabalan commented on code in PR #4739:
URL: https://github.com/apache/hudi/pull/4739#discussion_r868188701
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java:
##########
@@ -1301,4 +1359,33 @@ public void close() {
this.heartbeatClient.stop();
this.txnManager.close();
}
+
+ private void setWriteTimer(HoodieTable<T, I, K, O> table) {
+ String commitType = table.getMetaClient().getCommitActionType();
+ if (commitType.equals(HoodieTimeline.COMMIT_ACTION)) {
+ writeTimer = metrics.getCommitCtx();
+ } else if (commitType.equals(HoodieTimeline.DELTA_COMMIT_ACTION)) {
+ writeTimer = metrics.getDeltaCommitCtx();
+ }
+ }
+
+ private void tryUpgrade(HoodieTableMetaClient metaClient, Option<String>
instantTime) {
+ UpgradeDowngrade upgradeDowngrade =
+ new UpgradeDowngrade(metaClient, config, context,
upgradeDowngradeHelper);
+
+ if
(upgradeDowngrade.needsUpgradeOrDowngrade(HoodieTableVersion.current())) {
+ // Ensure no inflight commits by setting EAGER policy and explicitly
cleaning all failed commits
+ List<String> instantsToRollback = getInstantsToRollback(metaClient,
HoodieFailedWritesCleaningPolicy.EAGER, instantTime);
+
+ Map<String, Option<HoodiePendingRollbackInfo>> pendingRollbacks =
getPendingRollbackInfos(metaClient);
Review Comment:
this is how the code was in 0.9.0
```
if
(config.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()) {
this.txnManager.beginTransaction();
try {
// Ensure no inflight commits by setting EAGER policy and
explicitly cleaning all failed commits
this.rollbackFailedWrites(getInstantsToRollback(metaClient,
HoodieFailedWritesCleaningPolicy.EAGER));
new SparkUpgradeDowngrade(metaClient, config, context)
.run(metaClient, HoodieTableVersion.current(), config,
context, instantTime);
} finally {
this.txnManager.endTransaction();
}
} else {
upgradeDowngrade.run(metaClient, HoodieTableVersion.current(),
config, context, instantTime);
}
```
and here is how it looks in 0.10.0
```
if (upgradeDowngrade.needsUpgradeOrDowngrade(HoodieTableVersion.current())) {
// Ensure no inflight commits by setting EAGER policy and explicitly
cleaning all failed commits
List<String> instantsToRollback = getInstantsToRollback(
metaClient, HoodieFailedWritesCleaningPolicy.EAGER,
Option.of(instantTime));
Map<String, Option<HoodiePendingRollbackInfo>> pendingRollbacks =
getPendingRollbackInfos(metaClient);
instantsToRollback.forEach(entry ->
pendingRollbacks.putIfAbsent(entry, Option.empty()));
this.rollbackFailedWrites(pendingRollbacks, true);
new UpgradeDowngrade(
metaClient, config, context,
SparkUpgradeDowngradeHelper.getInstance())
.run(HoodieTableVersion.current(), instantTime);
metaClient.reloadActiveTimeline();
initializeMetadataTable(Option.of(instantTime));
}
```
So, atleast from 0.9.0, its evident that we were adding this only for
multi-writer scenario. i.e when someone migrates from a single writer to
multi-writer, we just wanted to ensure we rollback any partially failed
commits.
But I don't see any gaps as such w/ current state of things. bcoz, this code
gets exercised only when an upgrade is required. So, as per the guideline, we
should have only only writer in flight when an upgrade happens. So, we can do
eager rollbacks irrespective of whether multi-writer is enabled or not.
Let me know if you can think of a scenario where we can't trigger eager
rollbacks during upgrade ?
--
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]
