From 1d626506050b4cb9f83957dbcae501b9203a1abe Mon Sep 17 00:00:00 2001
From: jtian <jtian@netflix.com>
Date: Mon, 19 Jul 2021 16:06:44 -0700
Subject: [PATCH 1/2] Publish the list of partition ids on
 TABLE_PARTITION_UPDATE event

---
 .../TablePartitionsUpdatePayload.java         |  7 +++++-
 .../sns/SNSNotificationServiceImpl.java       |  6 +++--
 .../sns/SNSNotificationServiceUtil.java       | 23 +++++++++++++++++--
 3 files changed, 31 insertions(+), 5 deletions(-)

diff --git a/metacat-common/src/main/java/com/netflix/metacat/common/dto/notifications/sns/payloads/TablePartitionsUpdatePayload.java b/metacat-common/src/main/java/com/netflix/metacat/common/dto/notifications/sns/payloads/TablePartitionsUpdatePayload.java
index 7a83d7733..0eb9fabea 100644
--- a/metacat-common/src/main/java/com/netflix/metacat/common/dto/notifications/sns/payloads/TablePartitionsUpdatePayload.java
+++ b/metacat-common/src/main/java/com/netflix/metacat/common/dto/notifications/sns/payloads/TablePartitionsUpdatePayload.java
@@ -24,6 +24,7 @@
 import lombok.ToString;
 
 import javax.annotation.Nullable;
+import java.util.List;
 
 /**
  * Information about how the partitions have changed when a table was updated.
@@ -39,6 +40,7 @@ public class TablePartitionsUpdatePayload {
     private final int numCreatedPartitions;
     private final int numDeletedPartitions;
     private final String message;
+    private final List<String> partitionsUpdated;
 
     /**
      * Constructor.
@@ -46,16 +48,19 @@ public class TablePartitionsUpdatePayload {
      * @param numCreatedPartitions The number of partitions that were created for the table
      * @param numDeletedPartitions The number of partitions that were deleted from the table
      * @param message The message about the partition ids.
+     * @param partitionsUpdated The list of ids of the partitions that were updated
      */
     @JsonCreator
     public TablePartitionsUpdatePayload(
         @Nullable @JsonProperty("latestDeleteColumnValue") final String latestDeleteColumnValue,
         @JsonProperty("numCreatedPartitions") final int numCreatedPartitions,
         @JsonProperty("numDeletedPartitions") final int numDeletedPartitions,
-        @JsonProperty("message") final String message) {
+        @JsonProperty("message") final String message,
+        @JsonProperty("partitionsUpdated") final List<String> partitionsUpdated) {
         this.latestDeleteColumnValue = latestDeleteColumnValue;
         this.numCreatedPartitions = numCreatedPartitions;
         this.numDeletedPartitions = numDeletedPartitions;
         this.message = message;
+        this.partitionsUpdated = partitionsUpdated;
     }
 }
diff --git a/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceImpl.java b/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceImpl.java
index 233e7b598..3142ae0f0 100644
--- a/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceImpl.java
+++ b/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceImpl.java
@@ -123,7 +123,8 @@ public void notifyOfPartitionAddition(final MetacatSaveTablePartitionPostEvent e
                 createTablePartitionsUpdatePayload(event.getPartitions(), event);
         } else {
             partitionsUpdatePayload = new TablePartitionsUpdatePayload(null, event.getPartitions().size(), 0,
-                SNSNotificationPartitionAddMsg.PARTITION_KEY_UNABLED.name());
+                SNSNotificationPartitionAddMsg.PARTITION_KEY_UNABLED.name(),
+                SNSNotificationServiceUtil.getPartitionNameListFromDtos(event.getPartitions()));
         }
         final UpdateTablePartitionsMessage tableMessage = new UpdateTablePartitionsMessage(
             UUID.randomUUID().toString(),
@@ -182,7 +183,8 @@ public void notifyOfPartitionDeletion(final MetacatDeleteTablePartitionPostEvent
         final String requestId = event.getRequestContext().getId();
         final TablePartitionsUpdatePayload partitionsUpdatePayload;
         partitionsUpdatePayload = new TablePartitionsUpdatePayload(null, 0, event.getPartitions().size(),
-            SNSNotificationPartitionAddMsg.PARTITION_KEY_UNABLED.name());
+            SNSNotificationPartitionAddMsg.PARTITION_KEY_UNABLED.name(),
+            SNSNotificationServiceUtil.getPartitionNameListFromDtos(event.getPartitions()));
 
         final UpdateTablePartitionsMessage tableMessage = new UpdateTablePartitionsMessage(
             UUID.randomUUID().toString(),
diff --git a/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceUtil.java b/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceUtil.java
index a4c1ac6ac..44b4fa6c1 100644
--- a/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceUtil.java
+++ b/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceUtil.java
@@ -60,6 +60,7 @@ public final class SNSNotificationServiceUtil {
     //ISO basic date format: 20180101
     private static final Pattern TIMESTAMP_FORMAT = Pattern.compile("^(?<time>\\d{10})(?:\\d{3})?(?:\\.\\d+)?$");
     private static final Pattern ISO_BASIC = Pattern.compile("^\\d{8}$");
+    private static final int PARTS_UPDATED_LIST_MAX_SIZE = 1000;
     private UserMetadataService userMetadataService;
     private final DateFormat simpleDateFormatRegional = new SimpleDateFormat("yyyyMMdd");
     private final DateFormat simpleDateFormatUTC = new SimpleDateFormat("yyyyMMdd");
@@ -111,9 +112,11 @@ public TablePartitionsUpdatePayload createTablePartitionsUpdatePayload(
                         null,
                         partitionDtos.size(),
                         0,
-                        SNSNotificationPartitionAddMsg.EMPTY_DELETE_COLUMN.name()
+                        SNSNotificationPartitionAddMsg.EMPTY_DELETE_COLUMN.name(),
+                        getPartitionNameListFromDtos(partitionDtos)
                     );
                 }
+
                 deleteColumnValues = getSortedDeletionPartitionKeys(partitionDtos, deleteColumn);
                 //Calculate the latest partition key from candidates
                 if (deleteColumnValues != null && !deleteColumnValues.isEmpty()) {
@@ -153,7 +156,8 @@ public TablePartitionsUpdatePayload createTablePartitionsUpdatePayload(
             latestDeleteColumnValue,
             partitionDtos.size(),
             0,
-            message
+            message,
+            getPartitionNameListFromDtos(partitionDtos)
         );
     }
 
@@ -174,6 +178,21 @@ private static List<String> getSortedDeletionPartitionKeys(final List<PartitionD
             .collect(Collectors.toList());
     }
 
+    /**
+     * get partition name list from list of partitionDtos
+     *
+     * @param partitionDtos partition DTOs
+     * @return descending order deletion column
+     */
+    @VisibleForTesting
+    protected static List<String> getPartitionNameListFromDtos(final List<PartitionDto> partitionDtos) {
+        return partitionDtos.stream()
+            .limit(PARTS_UPDATED_LIST_MAX_SIZE)
+            .map(PartitionDto::getName)
+            .map(QualifiedName::getPartitionName)
+            .collect(Collectors.toList());
+    }
+
     /**
      * convert string to time stamp.
      * Three formats are accepted for now, which are basic standard ISO format and epoch timestamp format.

From a9858f0ed4cdfe4589e7259d38b72d74e69209b2 Mon Sep 17 00:00:00 2001
From: jtian <jtian@netflix.com>
Date: Mon, 19 Jul 2021 16:06:44 -0700
Subject: [PATCH 2/2] Publish the list of partition ids on
 TABLE_PARTITION_UPDATE event

---
 .../sns/SNSMessageFactorySpec.groovy           |  3 ++-
 .../sns/SNSNotificationServiceUtil.java        | 18 +++++++++++-------
 2 files changed, 13 insertions(+), 8 deletions(-)

diff --git a/metacat-common/src/test/groovy/com/netflix/metacat/common/dto/notifications/sns/SNSMessageFactorySpec.groovy b/metacat-common/src/test/groovy/com/netflix/metacat/common/dto/notifications/sns/SNSMessageFactorySpec.groovy
index 620fabea0..d0c4a312b 100644
--- a/metacat-common/src/test/groovy/com/netflix/metacat/common/dto/notifications/sns/SNSMessageFactorySpec.groovy
+++ b/metacat-common/src/test/groovy/com/netflix/metacat/common/dto/notifications/sns/SNSMessageFactorySpec.groovy
@@ -111,7 +111,8 @@ class SNSMessageFactorySpec extends Specification {
                     UUID.randomUUID().toString(),
                     10,
                     15,
-                    ""
+                    "",
+                    Arrays.asList(UUID.randomUUID().toString())
                 )
             )
         )       | UpdateTablePartitionsMessage.class
diff --git a/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceUtil.java b/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceUtil.java
index 44b4fa6c1..706a8855f 100644
--- a/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceUtil.java
+++ b/metacat-main/src/main/java/com/netflix/metacat/main/services/notifications/sns/SNSNotificationServiceUtil.java
@@ -33,6 +33,7 @@
 import java.text.SimpleDateFormat;
 import java.time.Instant;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
@@ -60,7 +61,7 @@ public final class SNSNotificationServiceUtil {
     //ISO basic date format: 20180101
     private static final Pattern TIMESTAMP_FORMAT = Pattern.compile("^(?<time>\\d{10})(?:\\d{3})?(?:\\.\\d+)?$");
     private static final Pattern ISO_BASIC = Pattern.compile("^\\d{8}$");
-    private static final int PARTS_UPDATED_LIST_MAX_SIZE = 1000;
+    private static final int PARTITIONS_UPDATED_LIST_MAX_SIZE = 1000;
     private UserMetadataService userMetadataService;
     private final DateFormat simpleDateFormatRegional = new SimpleDateFormat("yyyyMMdd");
     private final DateFormat simpleDateFormatUTC = new SimpleDateFormat("yyyyMMdd");
@@ -179,17 +180,20 @@ private static List<String> getSortedDeletionPartitionKeys(final List<PartitionD
     }
 
     /**
-     * get partition name list from list of partitionDtos
+     * get partition name list from list of partitionDtos. The returned list is capped at
+     * the first PARTITIONS_UPDATED_LIST_MAX_SIZE elements, if there are more than that number of elements
+     * in the input then the return is empty which serves as a signal that complete list cannot be included
      *
      * @param partitionDtos partition DTOs
-     * @return descending order deletion column
+     * @return list of partition ids from the input list
      */
-    @VisibleForTesting
     protected static List<String> getPartitionNameListFromDtos(final List<PartitionDto> partitionDtos) {
+        if (partitionDtos.size() > PARTITIONS_UPDATED_LIST_MAX_SIZE) {
+            // empty list signals
+            return Collections.emptyList();
+        }
         return partitionDtos.stream()
-            .limit(PARTS_UPDATED_LIST_MAX_SIZE)
-            .map(PartitionDto::getName)
-            .map(QualifiedName::getPartitionName)
+            .map(dto -> dto.getName().getPartitionName())
             .collect(Collectors.toList());
     }