From 8203a086f8adc161405662850cb1ea60945be99f Mon Sep 17 00:00:00 2001
From: Swaranga <114035979+swaranga-netflix@users.noreply.github.com>
Date: Fri, 3 Feb 2023 14:55:42 -0800
Subject: [PATCH] Fix bug in owner selection when owner is present in serde
 (#529)

---
 .../main/services/impl/TableServiceImpl.java        | 13 +++++++++++++
 .../main/services/impl/TableServiceImplSpec.groovy  |  1 +
 2 files changed, 14 insertions(+)

diff --git a/metacat-main/src/main/java/com/netflix/metacat/main/services/impl/TableServiceImpl.java b/metacat-main/src/main/java/com/netflix/metacat/main/services/impl/TableServiceImpl.java
index a4bc61ae5..8403dd188 100644
--- a/metacat-main/src/main/java/com/netflix/metacat/main/services/impl/TableServiceImpl.java
+++ b/metacat-main/src/main/java/com/netflix/metacat/main/services/impl/TableServiceImpl.java
@@ -170,6 +170,18 @@ private void logOwnershipDiagnosticDetails(final QualifiedName name, final Table
 
                 log.info("Create table with invalid owner: {}. name: {}, table-dto: {}, context: {}, headers: {}",
                     tableOwner, name, tableDto, MetacatContextManager.getContext(), getHttpHeaders());
+            } else if (!isOwnerValid(MetacatContextManager.getContext().getUserName())) {
+                registry.counter(
+                    "unauth.user.createTable.requestContext",
+                    "catalog", name.getCatalogName(),
+                    "database", name.getDatabaseName(),
+                    "owner", StringUtils.isBlank(MetacatContextManager.getContext().getUserName()) ? "null" : tableOwner
+                ).increment();
+
+                log.info("Create table called with invalid owner: {} in request context."
+                             + " name: {}, table-dto: {}, context: {}, headers: {}",
+                    MetacatContextManager.getContext().getUserName(), name, tableDto,
+                    MetacatContextManager.getContext(), getHttpHeaders());
             }
         } catch (Exception ex) {
             log.warn("Error when logging diagnostic data for invalid owner table creation. name: {}, table: {}",
@@ -247,6 +259,7 @@ private void setOwnerIfNull(final TableDto tableDto) {
         final String serdeOwner = tableDto.getSerde().getOwner();
         if (isOwnerValid(serdeOwner)) {
             updateTableOwner(tableDto, serdeOwner);
+            return;
         }
 
         // At this point, if we still have not found a valid user,
diff --git a/metacat-main/src/test/groovy/com/netflix/metacat/main/services/impl/TableServiceImplSpec.groovy b/metacat-main/src/test/groovy/com/netflix/metacat/main/services/impl/TableServiceImplSpec.groovy
index 043350acf..6d5b48306 100644
--- a/metacat-main/src/test/groovy/com/netflix/metacat/main/services/impl/TableServiceImplSpec.groovy
+++ b/metacat-main/src/test/groovy/com/netflix/metacat/main/services/impl/TableServiceImplSpec.groovy
@@ -347,6 +347,7 @@ class TableServiceImplSpec extends Specification {
         initialDefinitionMetadata              | sessionUser | initialSerde                      || expectedDefMetadata                     | expectedSerde
         null                                   | null        | null                              || "{}"                                    | new StorageDto()
         null                                   | "ssarma"    | null                              || "{\"owner\":{\"userId\":\"ssarma\"}}"   | new StorageDto()
+        null                                   | "root"      | new StorageDto(owner: "swaranga") || "{\"owner\":{\"userId\":\"swaranga\"}}" | new StorageDto(owner: "swaranga")
         "{\"owner\":{\"userId\":\"ssarma\"}}"  | "asdf"      | new StorageDto(owner: "swaranga") || "{\"owner\":{\"userId\":\"ssarma\"}}"   | new StorageDto(owner: "swaranga")
         "{\"owner\":{\"userId\":\"metacat\"}}" | "ssarma"    | new StorageDto(owner: "swaranga") || "{\"owner\":{\"userId\":\"ssarma\"}}"   | new StorageDto(owner: "swaranga")
         "{\"owner\":{\"userId\":\"root\"}}"    | "ssarma"    | new StorageDto(owner: "swaranga") || "{\"owner\":{\"userId\":\"ssarma\"}}"   | new StorageDto(owner: "swaranga")