codope commented on code in PR #10807:
URL: https://github.com/apache/hudi/pull/10807#discussion_r1516523304
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimeGeneratorBase.java:
##########
@@ -79,24 +71,26 @@ public TimeGeneratorBase(HoodieTimeGeneratorConfig config,
SerializableConfigura
this.lockConfiguration = config.getLockConfiguration();
this.hadoopConf = hadoopConf;
- maxRetries =
lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY,
+ // The maximum times to retry in case there are failures.
+ int maxRetries =
lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY,
Integer.parseInt(DEFAULT_LOCK_ACQUIRE_NUM_RETRIES));
lockAcquireWaitTimeInMs =
lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY,
DEFAULT_LOCK_ACQUIRE_WAIT_TIMEOUT_MS);
- maxWaitTimeInMs =
lockConfiguration.getConfig().getLong(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY,
+ // The maximum time to wait for each time generation to resolve the clock
skew issue on distributed hosts.
+ long maxWaitTimeInMs =
lockConfiguration.getConfig().getLong(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY,
Long.parseLong(DEFAULT_LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS));
lockRetryHelper = new RetryHelper<>(maxWaitTimeInMs, maxRetries,
maxWaitTimeInMs,
Arrays.asList(HoodieLockException.class, InterruptedException.class),
"acquire timeGenerator lock");
}
- protected LockProvider getLockProvider() {
+ protected LockProvider<?> getLockProvider() {
Review Comment:
why this change here? maybe do in a separate commit?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java:
##########
@@ -581,4 +596,43 @@ public HoodieDefaultTimeline
mergeTimeline(HoodieDefaultTimeline timeline) {
};
return new HoodieDefaultTimeline(instantStream, details);
}
+
+ /**
+ * Computes the timeline hash and returns.
+ */
+ private String computeTimelineHash(List<HoodieInstant> instants) {
+ final MessageDigest md;
+ try {
+ md = MessageDigest.getInstance(HASHING_ALGORITHM);
+ instants.forEach(i -> md
+ .update(getUTF8Bytes(StringUtils.joinUsingDelim("_",
i.getTimestamp(), i.getAction(), i.getState().name()))));
+ } catch (NoSuchAlgorithmException nse) {
+ throw new HoodieException(nse);
+ }
+ return StringUtils.toHexString(md.digest());
+ }
+
+ /**
+ * Merges the given instant list into one and keep the sequence.
+ */
+ private static List<HoodieInstant> mergeInstants(List<HoodieInstant>
instants1, List<HoodieInstant> instants2) {
+ ValidationUtils.checkArgument(!instants1.isEmpty() &&
!instants2.isEmpty(), "The instants to merge can not be empty");
+ final boolean firstInstantListHappensEarlier =
HoodieTimeline.compareTimestamps(instants1.get(0).getTimestamp(),
LESSER_THAN_OR_EQUALS, instants2.get(0).getTimestamp());
+ // some optimization based on the assumption all the instant list is
already sorted.
+ // skip when one list contains all the instants from the other one.
+ final List<HoodieInstant> merged;
+ if (HoodieTimeline.compareTimestamps(instants1.get(instants1.size() -
1).getTimestamp(), LESSER_THAN_OR_EQUALS, instants2.get(0).getTimestamp())) {
+ merged = new ArrayList<>(instants1);
+ merged.addAll(instants2);
+ } else if (HoodieTimeline.compareTimestamps(instants2.get(instants2.size()
- 1).getTimestamp(), LESSER_THAN_OR_EQUALS, instants1.get(0).getTimestamp())) {
+ merged = new ArrayList<>(instants2);
+ merged.addAll(instants1);
+ } else {
+ merged = new ArrayList<>(instants1);
+ merged.addAll(instants2);
+ // sort the instants explicitly
+ Collections.sort(merged);
Review Comment:
should sorting be done outside if-else i.e. just before returning merged
list?
--
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]
