[flink] branch release-1.16 updated: [FLINK-28890][table] Fix semantic of latestLoadTime in caching lookup function

[renqs]

This is an automated email from the ASF dual-hosted git repository.

renqs pushed a commit to branch release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.16 by this push:
     new 0830c2ac819 [FLINK-28890][table] Fix semantic of latestLoadTime in 
caching lookup function
0830c2ac819 is described below

commit 0830c2ac819fd33628d11315b2485a12c23af5e6
Author: Qingsheng Ren <renqs...@gmail.com>
AuthorDate: Tue Aug 9 18:29:01 2022 +0800

    [FLINK-28890][table] Fix semantic of latestLoadTime in caching lookup 
function
---
 .../runtime/functions/table/lookup/CachingAsyncLookupFunction.java | 7 ++++---
 .../runtime/functions/table/lookup/CachingLookupFunction.java      | 7 ++++---
 2 files changed, 8 insertions(+), 6 deletions(-)

diff --git 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingAsyncLookupFunction.java
 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingAsyncLookupFunction.java
index c087ef5ef9f..9fdea1e0d13 100644
--- 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingAsyncLookupFunction.java
+++ 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingAsyncLookupFunction.java
@@ -88,6 +88,7 @@ public class CachingAsyncLookupFunction extends 
AsyncLookupFunction {
         if (cachedValues != null) {
             return CompletableFuture.completedFuture(cachedValues);
         } else {
+            long loadStartTime = System.currentTimeMillis();
             return delegate.asyncLookup(keyRow)
                     .whenComplete(
                             (lookupValues, throwable) -> {
@@ -99,8 +100,8 @@ public class CachingAsyncLookupFunction extends 
AsyncLookupFunction {
                                             String.format("Failed to lookup 
key '%s'", keyRow),
                                             throwable);
                                 }
+                                
updateLatestLoadTime(System.currentTimeMillis() - loadStartTime);
                                 loadCounter.inc();
-                                updateLatestLoadTime();
                                 Collection<RowData> cachingValues = 
lookupValues;
                                 if (lookupValues == null || 
lookupValues.isEmpty()) {
                                     cachingValues = Collections.emptyList();
@@ -124,10 +125,10 @@ public class CachingAsyncLookupFunction extends 
AsyncLookupFunction {
     }
 
     // --------------------------------- Helper functions 
----------------------------
-    private void updateLatestLoadTime() {
+    private synchronized void updateLatestLoadTime(long loadTime) {
         if (latestLoadTime == UNINITIALIZED) {
             cacheMetricGroup.latestLoadTimeGauge(() -> latestLoadTime);
         }
-        latestLoadTime = System.currentTimeMillis();
+        latestLoadTime = loadTime;
     }
 }
diff --git 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingLookupFunction.java
 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingLookupFunction.java
index b75dd035d94..fb0c3075685 100644
--- 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingLookupFunction.java
+++ 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/CachingLookupFunction.java
@@ -159,9 +159,10 @@ public class CachingLookupFunction extends LookupFunction {
             Preconditions.checkState(
                     delegate != null,
                     "User's lookup function can't be null, if there are 
possible cache misses.");
+            long loadStart = System.currentTimeMillis();
             Collection<RowData> lookupValues = delegate.lookup(keyRow);
+            updateLatestLoadTime(System.currentTimeMillis() - loadStart);
             loadCounter.inc();
-            updateLatestLoadTime();
             return lookupValues;
         } catch (Exception e) {
             // TODO: Should implement retry on failure logic as proposed in 
FLIP-234
@@ -170,7 +171,7 @@ public class CachingLookupFunction extends LookupFunction {
         }
     }
 
-    private void updateLatestLoadTime() {
+    private void updateLatestLoadTime(long loadTime) {
         checkNotNull(
                 cacheMetricGroup,
                 "Could not register metric '%s' as cache metric group is not 
initialized",
@@ -179,6 +180,6 @@ public class CachingLookupFunction extends LookupFunction {
         if (latestLoadTime == UNINITIALIZED) {
             cacheMetricGroup.latestLoadTimeGauge(() -> latestLoadTime);
         }
-        latestLoadTime = System.currentTimeMillis();
+        latestLoadTime = loadTime;
     }
 }