Retry definition metadata updates for vtts updates

Author: insyncoss

metacat-common-server/src/main/java/com/netflix/metacat/common/server/util/MetacatUtils.java

@@ -32,6 +32,8 @@ public class MetacatUtils {
 
     public static final String ICEBERG_MIGRATION_DO_NOT_MODIFY_TAG = "iceberg_migration_do_not_modify";
     public static final String NAME_TAGS = "tags";
+    private static final String DATA_DEPENDENCY_NODE = "data_dependency";
+    private static final String VTTS_FIELD = "valid_thru_utc_ts";
 
     /**
      * Iceberg common view field names.
@@ -128,6 +130,21 @@ public static Set<String> getTableTags(@Nullable final ObjectNode definitionMeta
         return tags;
     }
 
+    public static long getVtts(final ObjectNode metadata) {
+        if (metadata == null) {
+            return -1;
+        }
+        JsonNode tmp = metadata.get(DATA_DEPENDENCY_NODE);
+        if (tmp == null || !tmp.isObject()) {
+            return -1;
+        }
+        JsonNode vttsNode = tmp.get(VTTS_FIELD);
+        if (vttsNode == null || !vttsNode.canConvertToLong()) {
+            return -1;
+        }
+        return vttsNode.asLong();
+    }
+
     public static String getIcebergMigrationExceptionMsg(final String requestType,
                                                          final String tableName) {
         return String.format("%s to hive table: %s are temporarily blocked " +

metacat-main/src/main/java/com/netflix/metacat/main/services/impl/TableServiceImpl.java

@@ -684,6 +684,19 @@ public TableDto updateAndReturn(final QualifiedName name, final TableDto tableDt
                 log.info("Saving user metadata for table {}", name);
                 final long start = registry.clock().wallTime();
                 userMetadataService.saveMetadata(metacatRequestContext.getUserName(), tableDto, true);
+                final long vtts = MetacatUtils.getVtts(tableDto.getDefinitionMetadata());
+                if (vtts > 0) {
+                    log.info("Received vtts update for {} to {}", name, vtts);
+                    final Optional<ObjectNode> metadata = userMetadataService.getDefinitionMetadata(tableDto.getName());
+                    if (metadata.isPresent()) {
+                        final long currVtts = MetacatUtils.getVtts(metadata.get());
+                        if (currVtts < vtts) {
+                            log.info("Retrying vtts update for {} from {} to {}", name, currVtts, vtts);
+                            userMetadataService.saveMetadata(metacatRequestContext.getUserName(), tableDto, true);
+                        }
+                    }
+                }
+
                 final long duration = registry.clock().wallTime() - start;
                 log.info("Time taken to save user metadata for table {} is {} ms", name, duration);
                 registry.timer(registry.createId(Metrics.TimerSaveTableMetadata.getMetricName()).withTags(name.parts()))