You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2022/02/22 07:54:51 UTC
[hudi] branch master updated: [HUDI-3461] The archived timeline for flink streaming reader should not be reused (#4861)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 4affdd0 [HUDI-3461] The archived timeline for flink streaming reader should not be reused (#4861)
4affdd0 is described below
commit 4affdd0c8f02ccca4705515dcb6492c199e2cede
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Tue Feb 22 15:54:29 2022 +0800
[HUDI-3461] The archived timeline for flink streaming reader should not be reused (#4861)
* Before the patch, the flink streaming reader caches the meta client thus the archived timeline,
when fetching the instant details from the reused timeline, the exception throws
* Add a method in HoodieTableMetaClient to return a fresh new archived timeline each time
---
.../hudi/common/table/HoodieTableMetaClient.java | 18 ++++++++++--
.../table/timeline/HoodieArchivedTimeline.java | 34 +++++++++++++++++++---
.../apache/hudi/source/IncrementalInputSplits.java | 16 ++--------
3 files changed, 49 insertions(+), 19 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
index 740d569..4c1eac7 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
@@ -156,7 +156,7 @@ public class HoodieTableMetaClient implements Serializable {
*/
private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
in.defaultReadObject();
- fs = null; // will be lazily inited
+ fs = null; // will be lazily initialized
}
private void writeObject(java.io.ObjectOutputStream out) throws IOException {
@@ -330,7 +330,7 @@ public class HoodieTableMetaClient implements Serializable {
* Get the archived commits as a timeline. This is costly operation, as all data from the archived files are read.
* This should not be used, unless for historical debugging purposes.
*
- * @return Active commit timeline
+ * @return Archived commit timeline
*/
public synchronized HoodieArchivedTimeline getArchivedTimeline() {
if (archivedTimeline == null) {
@@ -340,6 +340,20 @@ public class HoodieTableMetaClient implements Serializable {
}
/**
+ * Returns fresh new archived commits as a timeline from startTs (inclusive).
+ *
+ * <p>This is costly operation if really early endTs is specified.
+ * Be caution to use this only when the time range is short.
+ *
+ * <p>This method is not thread safe.
+ *
+ * @return Archived commit timeline
+ */
+ public HoodieArchivedTimeline getArchivedTimeline(String startTs) {
+ return new HoodieArchivedTimeline(this, startTs);
+ }
+
+ /**
* Validate table properties.
* @param properties Properties from writeConfig.
* @param operationType operation type to be executed.
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
index 5ad3fa7..29f1665 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
@@ -79,13 +79,13 @@ public class HoodieArchivedTimeline extends HoodieDefaultTimeline {
private static final String ACTION_TYPE_KEY = "actionType";
private static final String ACTION_STATE = "actionState";
private HoodieTableMetaClient metaClient;
- private Map<String, byte[]> readCommits = new HashMap<>();
+ private final Map<String, byte[]> readCommits = new HashMap<>();
private static final Logger LOG = LogManager.getLogger(HoodieArchivedTimeline.class);
/**
- * Loads instants between (startTs, endTs].
- * Note that there is no lazy loading, so this may not work if really long time range (endTs-startTs) is specified.
+ * Loads all the archived instants.
+ * Note that there is no lazy loading, so this may not work if the archived timeline range is really long.
* TBD: Should we enforce maximum time range?
*/
public HoodieArchivedTimeline(HoodieTableMetaClient metaClient) {
@@ -97,6 +97,19 @@ public class HoodieArchivedTimeline extends HoodieDefaultTimeline {
}
/**
+ * Loads completed instants from startTs(inclusive).
+ * Note that there is no lazy loading, so this may not work if really early startTs is specified.
+ */
+ public HoodieArchivedTimeline(HoodieTableMetaClient metaClient, String startTs) {
+ this.metaClient = metaClient;
+ setInstants(loadInstants(new StartTsFilter(startTs), true,
+ record -> HoodieInstant.State.COMPLETED.toString().equals(record.get(ACTION_STATE).toString())));
+ // multiple casts will make this lambda serializable -
+ // http://docs.oracle.com/javase/specs/jls/se8/html/jls-15.html#jls-15.16
+ this.details = (Function<HoodieInstant, Option<byte[]>> & Serializable) this::getInstantDetails;
+ }
+
+ /**
* For serialization and de-serialization only.
*
* @deprecated
@@ -300,6 +313,19 @@ public class HoodieArchivedTimeline extends HoodieDefaultTimeline {
}
}
+ private static class StartTsFilter extends TimeRangeFilter {
+ private final String startTs;
+
+ public StartTsFilter(String startTs) {
+ super(startTs, null); // endTs is never used
+ this.startTs = startTs;
+ }
+
+ public boolean isInRange(HoodieInstant instant) {
+ return HoodieTimeline.compareTimestamps(instant.getTimestamp(), GREATER_THAN_OR_EQUALS, startTs);
+ }
+ }
+
/**
* Sort files by reverse order of version suffix in file name.
*/
@@ -330,7 +356,7 @@ public class HoodieArchivedTimeline extends HoodieDefaultTimeline {
// filter in-memory instants
Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, REPLACE_COMMIT_ACTION);
return new HoodieDefaultTimeline(getInstants().filter(i ->
- readCommits.keySet().contains(i.getTimestamp()))
+ readCommits.containsKey(i.getTimestamp()))
.filter(s -> validActions.contains(s.getAction())), details);
}
}
diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java b/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
index 58c38ef..02e0e25 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
@@ -250,22 +250,12 @@ public class IncrementalInputSplits implements Serializable {
InstantRange instantRange,
HoodieTimeline commitTimeline,
String tableName) {
- if (instantRange == null || commitTimeline.isBeforeTimelineStarts(instantRange.getStartInstant())) {
- // read the archived metadata if:
- // 1. the start commit is 'earliest';
- // 2. the start instant is archived.
- HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
+ if (commitTimeline.isBeforeTimelineStarts(instantRange.getStartInstant())) {
+ // read the archived metadata if the start instant is archived.
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline(instantRange.getStartInstant());
HoodieTimeline archivedCompleteTimeline = archivedTimeline.getCommitsTimeline().filterCompletedInstants();
if (!archivedCompleteTimeline.empty()) {
- final String endTs = archivedCompleteTimeline.lastInstant().get().getTimestamp();
Stream<HoodieInstant> instantStream = archivedCompleteTimeline.getInstants();
- if (instantRange != null) {
- archivedTimeline.loadInstantDetailsInMemory(instantRange.getStartInstant(), endTs);
- instantStream = instantStream.filter(s -> HoodieTimeline.compareTimestamps(s.getTimestamp(), GREATER_THAN_OR_EQUALS, instantRange.getStartInstant()));
- } else {
- final String startTs = archivedCompleteTimeline.firstInstant().get().getTimestamp();
- archivedTimeline.loadInstantDetailsInMemory(startTs, endTs);
- }
return maySkipCompaction(instantStream)
.map(instant -> WriteProfiles.getCommitMetadata(tableName, path, instant, archivedTimeline)).collect(Collectors.toList());
}