[iceberg] branch master updated: Core: Fix NPE in update notification when table metadata cannot be loaded (#2552)

[blue]

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

blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/master by this push:
     new ba54f9c  Core: Fix NPE in update notification when table metadata 
cannot be loaded (#2552)
ba54f9c is described below

commit ba54f9cfe37b28d3ad675bdc50daee75cda1f3c1
Author: Yufei Gu <yu...@apache.org>
AuthorDate: Wed May 19 17:55:40 2021 -0700

    Core: Fix NPE in update notification when table metadata cannot be loaded 
(#2552)
---
 .../java/org/apache/iceberg/MergingSnapshotProducer.java  | 15 ++++++++++++++-
 core/src/main/java/org/apache/iceberg/TableMetadata.java  |  1 +
 2 files changed, 15 insertions(+), 1 deletion(-)

diff --git a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java 
b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
index 775ff1a..45e4326 100644
--- a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
+++ b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
@@ -40,6 +40,8 @@ import 
org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import static org.apache.iceberg.TableProperties.MANIFEST_MIN_MERGE_COUNT;
 import static 
org.apache.iceberg.TableProperties.MANIFEST_MIN_MERGE_COUNT_DEFAULT;
@@ -49,6 +51,8 @@ import static 
org.apache.iceberg.TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED
 import static 
org.apache.iceberg.TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED_DEFAULT;
 
 abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
+  private static final Logger LOG = 
LoggerFactory.getLogger(MergingSnapshotProducer.class);
+
   // data is only added in "append" and "overwrite" operations
   private static final Set<String> VALIDATE_ADDED_FILES_OPERATIONS =
       ImmutableSet.of(DataOperations.APPEND, DataOperations.OVERWRITE);
@@ -390,7 +394,16 @@ abstract class MergingSnapshotProducer<ThisT> extends 
SnapshotProducer<ThisT> {
   @Override
   public Object updateEvent() {
     long snapshotId = snapshotId();
-    long sequenceNumber = ops.refresh().snapshot(snapshotId).sequenceNumber();
+    Snapshot justSaved = ops.refresh().snapshot(snapshotId);
+    long sequenceNumber = TableMetadata.INVALID_SEQUENCE_NUMBER;
+    if (justSaved == null) {
+      // The snapshot just saved may not be present if the latest metadata 
couldn't be loaded due to eventual
+      // consistency problems in refresh.
+      LOG.warn("Failed to load committed snapshot: omitting sequence number 
from notifications");
+    } else {
+      sequenceNumber = justSaved.sequenceNumber();
+    }
+
     return new CreateSnapshotEvent(
         tableName,
         operation(),
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java 
b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 532519e..5703cc4 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -48,6 +48,7 @@ import org.apache.iceberg.util.PropertyUtil;
  */
 public class TableMetadata implements Serializable {
   static final long INITIAL_SEQUENCE_NUMBER = 0;
+  static final long INVALID_SEQUENCE_NUMBER = -1;
   static final int DEFAULT_TABLE_FORMAT_VERSION = 1;
   static final int SUPPORTED_TABLE_FORMAT_VERSION = 2;
   static final int INITIAL_SPEC_ID = 0;