vinothchandar commented on a change in pull request #4123:
URL: https://github.com/apache/hudi/pull/4123#discussion_r757652764
##########
File path:
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
##########
@@ -851,22 +903,28 @@ public Boolean rollbackFailedWrites() {
*/
public Boolean rollbackFailedWrites(boolean skipLocking) {
HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
- List<String> instantsToRollback =
getInstantsToRollback(table.getMetaClient(),
config.getFailedWritesCleanPolicy(),
- Option.empty());
- rollbackFailedWrites(instantsToRollback, skipLocking);
+ List<String> instantsToRollback =
getInstantsToRollback(table.getMetaClient(),
config.getFailedWritesCleanPolicy(), Option.empty());
+ Map<String, Option<HoodiePendingRollbackInfo>> pendingRollbacks =
getPendingRollbacks(table.getMetaClient(), Option.empty());
+ instantsToRollback.forEach(entry -> pendingRollbacks.putIfAbsent(entry,
Option.empty()));
+
+ HashMap<String, Option<HoodiePendingRollbackInfo>>
reverseSortedRollbackInstants = pendingRollbacks.entrySet()
+ .stream().sorted((i1, i2) -> i2.getKey().compareTo(i1.getKey()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (e1,
e2) -> e1, LinkedHashMap::new));
Review comment:
if it wont work unless you have a `LinkedHashMap` then lets have the
type as `LinkedHashMap`
##########
File path:
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
##########
@@ -584,22 +588,40 @@ public void restoreToSavepoint(String savepointTime) {
@Deprecated
public boolean rollback(final String commitInstantTime) throws
HoodieRollbackException {
- return rollback(commitInstantTime, false);
+ HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
+ Option<HoodiePendingRollbackInfo> pendingRollbackInfo =
getPendingRollback(table.getMetaClient(), commitInstantTime);
+ return rollback(commitInstantTime, pendingRollbackInfo,false);
Review comment:
nit: space after comma?
##########
File path:
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
##########
@@ -851,22 +903,28 @@ public Boolean rollbackFailedWrites() {
*/
public Boolean rollbackFailedWrites(boolean skipLocking) {
HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
- List<String> instantsToRollback =
getInstantsToRollback(table.getMetaClient(),
config.getFailedWritesCleanPolicy(),
- Option.empty());
- rollbackFailedWrites(instantsToRollback, skipLocking);
+ List<String> instantsToRollback =
getInstantsToRollback(table.getMetaClient(),
config.getFailedWritesCleanPolicy(), Option.empty());
+ Map<String, Option<HoodiePendingRollbackInfo>> pendingRollbacks =
getPendingRollbacks(table.getMetaClient(), Option.empty());
+ instantsToRollback.forEach(entry -> pendingRollbacks.putIfAbsent(entry,
Option.empty()));
+
+ HashMap<String, Option<HoodiePendingRollbackInfo>>
reverseSortedRollbackInstants = pendingRollbacks.entrySet()
+ .stream().sorted((i1, i2) -> i2.getKey().compareTo(i1.getKey()))
Review comment:
nit: is nt there a replacement for `(i1, i2) ->
i2.getKey().compareTo(i1.getKey())` , some standard comparator. ?
##########
File path:
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
##########
@@ -851,22 +903,28 @@ public Boolean rollbackFailedWrites() {
*/
public Boolean rollbackFailedWrites(boolean skipLocking) {
HoodieTable<T, I, K, O> table = createTable(config, hadoopConf);
- List<String> instantsToRollback =
getInstantsToRollback(table.getMetaClient(),
config.getFailedWritesCleanPolicy(),
- Option.empty());
- rollbackFailedWrites(instantsToRollback, skipLocking);
+ List<String> instantsToRollback =
getInstantsToRollback(table.getMetaClient(),
config.getFailedWritesCleanPolicy(), Option.empty());
+ Map<String, Option<HoodiePendingRollbackInfo>> pendingRollbacks =
getPendingRollbacks(table.getMetaClient(), Option.empty());
Review comment:
rename: `getPendingRollbackInfos()`
##########
File path:
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
##########
@@ -838,6 +860,36 @@ private HoodieTimeline
getInflightTimelineExcludeCompactionAndClustering(HoodieT
return inflightTimelineExcludeClusteringCommit;
}
+ private Option<HoodiePendingRollbackInfo>
getPendingRollback(HoodieTableMetaClient metaClient, String commitToRollback) {
+ Option<HoodiePendingRollbackInfo> pendingRollbackInfo =
getPendingRollbacks(metaClient,
Option.of(commitToRollback)).get(commitToRollback);
+ return pendingRollbackInfo != null ? pendingRollbackInfo : Option.empty();
+ }
+
+ /**
+ * Fetch map of pending commits to be rolledback to (Rollback Instant and
Rollback plan).
+ * @param metaClient instance of {@link HoodieTableMetaClient} to use.
+ * @param commitToRollback optional commit time to be rolledback.
+ * @return map of pending commits to be rolledback instants to Rollback
Instnat and Rollback plan Pair.
+ */
+ protected Map<String, Option<HoodiePendingRollbackInfo>>
getPendingRollbacks(HoodieTableMetaClient metaClient, Option<String>
commitToRollback) {
+ Stream<Pair<String, Option<HoodiePendingRollbackInfo>>>
pendingRollbacksStream =
metaClient.getActiveTimeline().filterPendingRollbackTimeline().getInstants().map(
+ entry -> {
+ try {
+ HoodieRollbackPlan rollbackPlan =
RollbackUtils.getRollbackPlan(metaClient, entry);
+ return
Pair.of(rollbackPlan.getInstantToRollback().getCommitTime(), Option.of(new
HoodiePendingRollbackInfo(entry, rollbackPlan)));
+ } catch (IOException e) {
+ throw new HoodieIOException("Fetching rollback plan failed for " +
entry, e);
+ }
+ }
+ );
+
+ if (commitToRollback.isPresent()) {
Review comment:
could n't we move this to the call? keep this method clean - single
purpose for each method
--
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]
