hudi-agent commented on code in PR #20239:
URL: https://github.com/apache/hudi/pull/20239#discussion_r4235357552
##########
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/TimelineService.java:
##########
@@ -134,6 +134,13 @@ public static class Config implements Serializable {
@Parameter(names = {"--enable-remote-partitioner"}, description = "Enable
remote partitioner")
public boolean enableRemotePartitioner = false;
+ @Builder.Default
+ @Parameter(names = {"--stale-client-timeline-cache-size"}, description =
+ "Number of client timeline hashes to remember for which a refresh has
already been performed. "
+ + "When many executors share a stale timeline, only the first
request triggers a reload; "
+ + "the rest are served from the already-synced view. Set to 0 to
disable.")
+ public int staleClientTimelineCacheSize = 10;
Review Comment:
🤖 non-blocking: `EmbeddedTimelineService.startServer` doesn't pass this
value in from `HoodieWriteConfig`, so on the embedded server, where the
production report came from, the cache always runs at size 10 and users can't
set it to 0 if it misbehaves; would it make sense to wire it through the way
the marker batch configs are?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java:
##########
@@ -764,22 +777,121 @@ private boolean isLocalViewBehind(Context ctx) {
*/
private boolean syncIfLocalViewBehind(Context ctx) {
String basePath =
ctx.queryParam(RemoteHoodieTableFileSystemView.BASEPATH_PARAM);
+ String timelineHashFromClient = getTimelineHashParam(ctx);
SyncableFileSystemView view = viewManager.getFileSystemView(basePath);
synchronized (view) {
+ HoodieTimeline localTimeline = view.getTimeline();
+ // The final check already accepts a trailing clean with an otherwise
matching timeline, so a
+ // reload would be discarded work. Checked before isLocalViewBehind,
which hashes the whole
+ // timeline, because this only inspects the last instant.
+ // Other extensions still need a sync: restore may have deleted their
extra instants.
+ if (!shouldThrowExceptionIfLocalViewBehind(localTimeline,
timelineHashFromClient)) {
+ return false;
+ }
+ // A reload already performed for this exact client timeline cannot
produce a different
+ // outcome while the server timeline is unchanged, so skip it for the
remaining stale clients.
+ if (hasAlreadySyncedFor(basePath, localTimeline,
timelineHashFromClient, getLastInstantTsParam(ctx))) {
+ return false;
+ }
if (isLocalViewBehind(ctx)) {
String lastKnownInstantFromClient = getLastInstantTsParam(ctx);
- HoodieTimeline localTimeline =
viewManager.getFileSystemView(basePath).getTimeline();
if (log.isInfoEnabled()) {
log.info("Syncing view as client passed last known instant {} as
last known instant but server has the following last instant on timeline: {}",
lastKnownInstantFromClient, localTimeline.lastInstant());
}
+ Option<HoodieInstant> lastInstantBeforeSync =
localTimeline.lastInstant();
view.sync();
+ recordSyncedClientTimeline(basePath, view.getTimeline(),
lastInstantBeforeSync, timelineHashFromClient,
+ lastKnownInstantFromClient);
return true;
}
}
return false;
}
+ /**
+ * Whether a reload has already been performed for this client timeline
and can be skipped.
+ *
+ * <p>Only a previously observed reload is treated as evidence; the client
hash itself is never
+ * interpreted as proof that the server is ahead. The client's last known
instant is re-checked
+ * on every hit so that a restore, which deletes completed instants from
storage and moves the
+ * server timeline backwards, drops the entry instead of suppressing a
reload that is now needed.
+ */
+ private boolean hasAlreadySyncedFor(String basePath, HoodieTimeline
localTimeline,
+ String timelineHashFromClient, String
lastKnownInstantFromClient) {
+ if (!isCacheableClientTimeline(lastKnownInstantFromClient)) {
+ return false;
+ }
+ Map<String, String> syncedHashes =
syncedClientTimelineHashes.get(basePath);
+ if (syncedHashes == null) {
+ return false;
+ }
+ String serverHashWhenSynced =
syncedHashes.get(cacheKey(timelineHashFromClient, lastKnownInstantFromClient));
+ // Only the reload observed against this exact server timeline is
evidence. Once the server
+ // timeline moves, for any reason, a reload may produce a different
result and must run again.
+ if (serverHashWhenSynced == null ||
!serverHashWhenSynced.equals(localTimeline.getTimelineHash())) {
+ return false;
+ }
+ if
(!localTimeline.containsOrBeforeTimelineStarts(lastKnownInstantFromClient)) {
+ syncedHashes.remove(cacheKey(timelineHashFromClient,
lastKnownInstantFromClient));
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * The refresh decision depends on the client's last known instant as well
as its timeline hash,
+ * so both identify a recorded reload.
+ */
+ private String cacheKey(String timelineHashFromClient, String
lastKnownInstantFromClient) {
+ return timelineHashFromClient + "_" + lastKnownInstantFromClient;
+ }
+
+ /**
+ * Whether a reload for this client is worth recording.
+ *
+ * <p>A client that sent no last known instant has not said where it
stands, which only happens
+ * before its timeline holds a completed instant. There are no file slices
to serve from a stale
+ * view at that point, so nothing is recorded rather than treating the
absent value as an
+ * identity shared by unrelated clients.
+ */
+ private boolean isCacheableClientTimeline(String
lastKnownInstantFromClient) {
+ return timelineServiceConfig.staleClientTimelineCacheSize > 0
+ &&
!HoodieTimeline.INVALID_INSTANT_TS.equals(lastKnownInstantFromClient)
+ && !StringUtils.isNullOrEmpty(lastKnownInstantFromClient);
+ }
+
+ /**
+ * Records that a reload was performed for this client timeline, so that
the remaining stale
+ * clients sharing it do not each trigger another one.
+ *
+ * <p>Nothing is recorded when the reload dropped instants, which
indicates the server timeline
+ * moved backwards rather than forwards and the previous records can no
longer be trusted.
+ */
+ private void recordSyncedClientTimeline(String basePath, HoodieTimeline
syncedTimeline,
+ Option<HoodieInstant>
lastInstantBeforeSync, String timelineHashFromClient,
+ String lastKnownInstantFromClient)
{
+ int cacheSize = timelineServiceConfig.staleClientTimelineCacheSize;
+ if (!isCacheableClientTimeline(lastKnownInstantFromClient)) {
+ return;
+ }
+ if (lastInstantBeforeSync.isPresent()
+ &&
!syncedTimeline.containsInstant(lastInstantBeforeSync.get().requestedTime())) {
+ // Instants were deleted from storage, e.g. by a restore. Earlier
records described a
+ // timeline that no longer exists, so discard them rather than reason
about what survived.
+ syncedClientTimelineHashes.remove(basePath);
+ return;
+ }
+ syncedClientTimelineHashes
+ .computeIfAbsent(basePath, ignored -> new LinkedHashMap<String,
String>(cacheSize + 1, 0.75f, false) {
Review Comment:
🤖 non-blocking: `refreshTable` and `unregisterBasePath` replace the view
object through `clearFileSystemView`, so a request still inside `synchronized
(oldView)` and a new one inside `synchronized (newView)` can both `put` into
this `LinkedHashMap`; could the inner map be wrapped in
`Collections.synchronizedMap(...)` (or locked on directly) so it doesn't rely
on the view monitor?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java:
##########
@@ -764,22 +777,121 @@ private boolean isLocalViewBehind(Context ctx) {
*/
private boolean syncIfLocalViewBehind(Context ctx) {
String basePath =
ctx.queryParam(RemoteHoodieTableFileSystemView.BASEPATH_PARAM);
+ String timelineHashFromClient = getTimelineHashParam(ctx);
SyncableFileSystemView view = viewManager.getFileSystemView(basePath);
synchronized (view) {
+ HoodieTimeline localTimeline = view.getTimeline();
+ // The final check already accepts a trailing clean with an otherwise
matching timeline, so a
+ // reload would be discarded work. Checked before isLocalViewBehind,
which hashes the whole
+ // timeline, because this only inspects the last instant.
+ // Other extensions still need a sync: restore may have deleted their
extra instants.
+ if (!shouldThrowExceptionIfLocalViewBehind(localTimeline,
timelineHashFromClient)) {
+ return false;
+ }
+ // A reload already performed for this exact client timeline cannot
produce a different
+ // outcome while the server timeline is unchanged, so skip it for the
remaining stale clients.
+ if (hasAlreadySyncedFor(basePath, localTimeline,
timelineHashFromClient, getLastInstantTsParam(ctx))) {
+ return false;
+ }
if (isLocalViewBehind(ctx)) {
String lastKnownInstantFromClient = getLastInstantTsParam(ctx);
- HoodieTimeline localTimeline =
viewManager.getFileSystemView(basePath).getTimeline();
if (log.isInfoEnabled()) {
log.info("Syncing view as client passed last known instant {} as
last known instant but server has the following last instant on timeline: {}",
lastKnownInstantFromClient, localTimeline.lastInstant());
}
+ Option<HoodieInstant> lastInstantBeforeSync =
localTimeline.lastInstant();
view.sync();
+ recordSyncedClientTimeline(basePath, view.getTimeline(),
lastInstantBeforeSync, timelineHashFromClient,
+ lastKnownInstantFromClient);
return true;
}
}
return false;
}
+ /**
+ * Whether a reload has already been performed for this client timeline
and can be skipped.
+ *
+ * <p>Only a previously observed reload is treated as evidence; the client
hash itself is never
+ * interpreted as proof that the server is ahead. The client's last known
instant is re-checked
+ * on every hit so that a restore, which deletes completed instants from
storage and moves the
+ * server timeline backwards, drops the entry instead of suppressing a
reload that is now needed.
+ */
+ private boolean hasAlreadySyncedFor(String basePath, HoodieTimeline
localTimeline,
+ String timelineHashFromClient, String
lastKnownInstantFromClient) {
+ if (!isCacheableClientTimeline(lastKnownInstantFromClient)) {
+ return false;
+ }
+ Map<String, String> syncedHashes =
syncedClientTimelineHashes.get(basePath);
+ if (syncedHashes == null) {
+ return false;
+ }
+ String serverHashWhenSynced =
syncedHashes.get(cacheKey(timelineHashFromClient, lastKnownInstantFromClient));
+ // Only the reload observed against this exact server timeline is
evidence. Once the server
+ // timeline moves, for any reason, a reload may produce a different
result and must run again.
+ if (serverHashWhenSynced == null ||
!serverHashWhenSynced.equals(localTimeline.getTimelineHash())) {
+ return false;
+ }
+ if
(!localTimeline.containsOrBeforeTimelineStarts(lastKnownInstantFromClient)) {
Review Comment:
🤖 non-blocking: this re-check only runs while the server timeline still
hashes to the recorded post-sync value, so it can't see a restore that reached
storage after that reload and really only covers a reload that never reached
the client's instant; could the javadoc above say so, since restore is actually
handled by the hash comparison and the clear in `recordSyncedClientTimeline`?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java:
##########
@@ -764,22 +777,121 @@ private boolean isLocalViewBehind(Context ctx) {
*/
private boolean syncIfLocalViewBehind(Context ctx) {
String basePath =
ctx.queryParam(RemoteHoodieTableFileSystemView.BASEPATH_PARAM);
+ String timelineHashFromClient = getTimelineHashParam(ctx);
SyncableFileSystemView view = viewManager.getFileSystemView(basePath);
synchronized (view) {
+ HoodieTimeline localTimeline = view.getTimeline();
+ // The final check already accepts a trailing clean with an otherwise
matching timeline, so a
+ // reload would be discarded work. Checked before isLocalViewBehind,
which hashes the whole
+ // timeline, because this only inspects the last instant.
+ // Other extensions still need a sync: restore may have deleted their
extra instants.
+ if (!shouldThrowExceptionIfLocalViewBehind(localTimeline,
timelineHashFromClient)) {
+ return false;
+ }
+ // A reload already performed for this exact client timeline cannot
produce a different
+ // outcome while the server timeline is unchanged, so skip it for the
remaining stale clients.
+ if (hasAlreadySyncedFor(basePath, localTimeline,
timelineHashFromClient, getLastInstantTsParam(ctx))) {
+ return false;
+ }
if (isLocalViewBehind(ctx)) {
String lastKnownInstantFromClient = getLastInstantTsParam(ctx);
- HoodieTimeline localTimeline =
viewManager.getFileSystemView(basePath).getTimeline();
if (log.isInfoEnabled()) {
log.info("Syncing view as client passed last known instant {} as
last known instant but server has the following last instant on timeline: {}",
lastKnownInstantFromClient, localTimeline.lastInstant());
}
+ Option<HoodieInstant> lastInstantBeforeSync =
localTimeline.lastInstant();
view.sync();
+ recordSyncedClientTimeline(basePath, view.getTimeline(),
lastInstantBeforeSync, timelineHashFromClient,
+ lastKnownInstantFromClient);
return true;
}
}
return false;
}
+ /**
+ * Whether a reload has already been performed for this client timeline
and can be skipped.
+ *
+ * <p>Only a previously observed reload is treated as evidence; the client
hash itself is never
+ * interpreted as proof that the server is ahead. The client's last known
instant is re-checked
+ * on every hit so that a restore, which deletes completed instants from
storage and moves the
+ * server timeline backwards, drops the entry instead of suppressing a
reload that is now needed.
+ */
+ private boolean hasAlreadySyncedFor(String basePath, HoodieTimeline
localTimeline,
+ String timelineHashFromClient, String
lastKnownInstantFromClient) {
+ if (!isCacheableClientTimeline(lastKnownInstantFromClient)) {
+ return false;
+ }
+ Map<String, String> syncedHashes =
syncedClientTimelineHashes.get(basePath);
+ if (syncedHashes == null) {
+ return false;
+ }
+ String serverHashWhenSynced =
syncedHashes.get(cacheKey(timelineHashFromClient, lastKnownInstantFromClient));
+ // Only the reload observed against this exact server timeline is
evidence. Once the server
+ // timeline moves, for any reason, a reload may produce a different
result and must run again.
+ if (serverHashWhenSynced == null ||
!serverHashWhenSynced.equals(localTimeline.getTimelineHash())) {
+ return false;
+ }
+ if
(!localTimeline.containsOrBeforeTimelineStarts(lastKnownInstantFromClient)) {
+ syncedHashes.remove(cacheKey(timelineHashFromClient,
lastKnownInstantFromClient));
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * The refresh decision depends on the client's last known instant as well
as its timeline hash,
+ * so both identify a recorded reload.
+ */
+ private String cacheKey(String timelineHashFromClient, String
lastKnownInstantFromClient) {
+ return timelineHashFromClient + "_" + lastKnownInstantFromClient;
+ }
+
+ /**
+ * Whether a reload for this client is worth recording.
+ *
+ * <p>A client that sent no last known instant has not said where it
stands, which only happens
+ * before its timeline holds a completed instant. There are no file slices
to serve from a stale
+ * view at that point, so nothing is recorded rather than treating the
absent value as an
+ * identity shared by unrelated clients.
+ */
+ private boolean isCacheableClientTimeline(String
lastKnownInstantFromClient) {
+ return timelineServiceConfig.staleClientTimelineCacheSize > 0
+ &&
!HoodieTimeline.INVALID_INSTANT_TS.equals(lastKnownInstantFromClient)
+ && !StringUtils.isNullOrEmpty(lastKnownInstantFromClient);
+ }
+
+ /**
+ * Records that a reload was performed for this client timeline, so that
the remaining stale
+ * clients sharing it do not each trigger another one.
+ *
+ * <p>Nothing is recorded when the reload dropped instants, which
indicates the server timeline
+ * moved backwards rather than forwards and the previous records can no
longer be trusted.
+ */
+ private void recordSyncedClientTimeline(String basePath, HoodieTimeline
syncedTimeline,
+ Option<HoodieInstant>
lastInstantBeforeSync, String timelineHashFromClient,
+ String lastKnownInstantFromClient)
{
+ int cacheSize = timelineServiceConfig.staleClientTimelineCacheSize;
+ if (!isCacheableClientTimeline(lastKnownInstantFromClient)) {
+ return;
Review Comment:
🤖 nit: could you move the LRU `LinkedHashMap` construction into a small
named helper (e.g. `newBoundedSyncedHashes(cacheSize)`)? The anonymous subclass
inside `computeIfAbsent` makes `recordSyncedClientTimeline` hard to scan.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java:
##########
@@ -764,22 +777,121 @@ private boolean isLocalViewBehind(Context ctx) {
*/
private boolean syncIfLocalViewBehind(Context ctx) {
String basePath =
ctx.queryParam(RemoteHoodieTableFileSystemView.BASEPATH_PARAM);
+ String timelineHashFromClient = getTimelineHashParam(ctx);
SyncableFileSystemView view = viewManager.getFileSystemView(basePath);
synchronized (view) {
+ HoodieTimeline localTimeline = view.getTimeline();
+ // The final check already accepts a trailing clean with an otherwise
matching timeline, so a
+ // reload would be discarded work. Checked before isLocalViewBehind,
which hashes the whole
+ // timeline, because this only inspects the last instant.
+ // Other extensions still need a sync: restore may have deleted their
extra instants.
+ if (!shouldThrowExceptionIfLocalViewBehind(localTimeline,
timelineHashFromClient)) {
+ return false;
+ }
+ // A reload already performed for this exact client timeline cannot
produce a different
+ // outcome while the server timeline is unchanged, so skip it for the
remaining stale clients.
+ if (hasAlreadySyncedFor(basePath, localTimeline,
timelineHashFromClient, getLastInstantTsParam(ctx))) {
+ return false;
+ }
if (isLocalViewBehind(ctx)) {
String lastKnownInstantFromClient = getLastInstantTsParam(ctx);
- HoodieTimeline localTimeline =
viewManager.getFileSystemView(basePath).getTimeline();
if (log.isInfoEnabled()) {
log.info("Syncing view as client passed last known instant {} as
last known instant but server has the following last instant on timeline: {}",
lastKnownInstantFromClient, localTimeline.lastInstant());
}
+ Option<HoodieInstant> lastInstantBeforeSync =
localTimeline.lastInstant();
view.sync();
+ recordSyncedClientTimeline(basePath, view.getTimeline(),
lastInstantBeforeSync, timelineHashFromClient,
+ lastKnownInstantFromClient);
return true;
}
}
return false;
}
+ /**
+ * Whether a reload has already been performed for this client timeline
and can be skipped.
+ *
+ * <p>Only a previously observed reload is treated as evidence; the client
hash itself is never
+ * interpreted as proof that the server is ahead. The client's last known
instant is re-checked
+ * on every hit so that a restore, which deletes completed instants from
storage and moves the
+ * server timeline backwards, drops the entry instead of suppressing a
reload that is now needed.
+ */
+ private boolean hasAlreadySyncedFor(String basePath, HoodieTimeline
localTimeline,
+ String timelineHashFromClient, String
lastKnownInstantFromClient) {
+ if (!isCacheableClientTimeline(lastKnownInstantFromClient)) {
+ return false;
+ }
+ Map<String, String> syncedHashes =
syncedClientTimelineHashes.get(basePath);
+ if (syncedHashes == null) {
+ return false;
+ }
+ String serverHashWhenSynced =
syncedHashes.get(cacheKey(timelineHashFromClient, lastKnownInstantFromClient));
+ // Only the reload observed against this exact server timeline is
evidence. Once the server
+ // timeline moves, for any reason, a reload may produce a different
result and must run again.
+ if (serverHashWhenSynced == null ||
!serverHashWhenSynced.equals(localTimeline.getTimelineHash())) {
+ return false;
+ }
+ if
(!localTimeline.containsOrBeforeTimelineStarts(lastKnownInstantFromClient)) {
+ syncedHashes.remove(cacheKey(timelineHashFromClient,
lastKnownInstantFromClient));
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * The refresh decision depends on the client's last known instant as well
as its timeline hash,
+ * so both identify a recorded reload.
+ */
+ private String cacheKey(String timelineHashFromClient, String
lastKnownInstantFromClient) {
+ return timelineHashFromClient + "_" + lastKnownInstantFromClient;
+ }
+
+ /**
+ * Whether a reload for this client is worth recording.
+ *
+ * <p>A client that sent no last known instant has not said where it
stands, which only happens
+ * before its timeline holds a completed instant. There are no file slices
to serve from a stale
+ * view at that point, so nothing is recorded rather than treating the
absent value as an
+ * identity shared by unrelated clients.
Review Comment:
🤖 nit: `isCacheableClientTimeline` also checks `staleClientTimelineCacheSize
> 0`, so the name reads as only about the client timeline; would
`isCachingEnabledFor(lastKnownInstantFromClient)` or splitting out the config
check be clearer?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
