fix(entity-service): handle no-op system-metadata batches

* revert previous changes
* disable legacy locking
* handle no-op when no change to system metadata

Author: david-leifker

metadata-io/src/main/java/com/linkedin/metadata/entity/EntityServiceImpl.java

@@ -1051,7 +1051,8 @@ This condition is specifically for an older conditional write ingestAspectIfNotP
                 }
               } else {
                 MetricUtils.counter(EntityServiceImpl.class, "batch_empty_transaction").inc();
-                log.warn("Empty transaction detected. {}", inputBatch);
+                // This includes no-op batches. i.e. patch removing non-existent items
+                log.debug("Empty transaction detected. {}", inputBatch);
               }
 
               return upsertResults;
@@ -2533,9 +2534,6 @@ private UpdateAspectResult ingestAspectToLocalDB(
         .setLastRunId(writeItem.getSystemMetadata().getRunId(GetMode.NULL), SetMode.IGNORE_NULL);
 
     // 2. Compare the latest existing and new.
-    final RecordTemplate databaseValue =
-        databaseAspect == null ? null : databaseAspect.getRecordTemplate();
-
     final EntityAspect.EntitySystemAspect previousBatchAspect =
         (EntityAspect.EntitySystemAspect) writeItem.getPreviousSystemAspect();
     final RecordTemplate previousValue =
@@ -2544,43 +2542,49 @@ private UpdateAspectResult ingestAspectToLocalDB(
     // 3. If there is no difference between existing and new, we just update
     // the lastObserved in system metadata. RunId should stay as the original runId
     if (previousValue != null
-        && DataTemplateUtil.areEqual(databaseValue, writeItem.getRecordTemplate())) {
+        && DataTemplateUtil.areEqual(previousValue, writeItem.getRecordTemplate())) {
 
       SystemMetadata latestSystemMetadata = previousBatchAspect.getSystemMetadata();
       latestSystemMetadata.setLastObserved(writeItem.getSystemMetadata().getLastObserved());
       latestSystemMetadata.setLastRunId(
           writeItem.getSystemMetadata().getLastRunId(GetMode.NULL), SetMode.IGNORE_NULL);
 
-      previousBatchAspect
-          .getEntityAspect()
-          .setSystemMetadata(RecordUtils.toJsonString(latestSystemMetadata));
-
-      log.info(
-          "Ingesting aspect with name {}, urn {}",
-          previousBatchAspect.getAspectName(),
-          previousBatchAspect.getUrn());
-      aspectDao.saveAspect(txContext, previousBatchAspect.getEntityAspect(), false);
-
-      // metrics
-      aspectDao.incrementWriteMetrics(
-          previousBatchAspect.getAspectName(),
-          1,
-          previousBatchAspect.getMetadataRaw().getBytes(StandardCharsets.UTF_8).length);
-
-      return UpdateAspectResult.builder()
-          .urn(writeItem.getUrn())
-          .oldValue(previousValue)
-          .newValue(previousValue)
-          .oldSystemMetadata(previousBatchAspect.getSystemMetadata())
-          .newSystemMetadata(latestSystemMetadata)
-          .operation(MetadataAuditOperation.UPDATE)
-          .auditStamp(writeItem.getAuditStamp())
-          .maxVersion(0)
-          .build();
+      if (!latestSystemMetadata.equals(previousBatchAspect.getSystemMetadata())) {
+
+        previousBatchAspect
+            .getEntityAspect()
+            .setSystemMetadata(RecordUtils.toJsonString(latestSystemMetadata));
+
+        log.debug(
+            "Update aspect with name {}, urn {}",
+            previousBatchAspect.getAspectName(),
+            previousBatchAspect.getUrn());
+        aspectDao.saveAspect(txContext, previousBatchAspect.getEntityAspect(), false);
+
+        // metrics
+        aspectDao.incrementWriteMetrics(
+            previousBatchAspect.getAspectName(),
+            1,
+            previousBatchAspect.getMetadataRaw().getBytes(StandardCharsets.UTF_8).length);
+
+        return UpdateAspectResult.builder()
+            .urn(writeItem.getUrn())
+            .oldValue(previousValue)
+            .newValue(previousValue)
+            .oldSystemMetadata(previousBatchAspect.getSystemMetadata())
+            .newSystemMetadata(latestSystemMetadata)
+            .operation(MetadataAuditOperation.UPDATE)
+            .auditStamp(writeItem.getAuditStamp())
+            .maxVersion(0)
+            .build();
+      } else {
+        MetricUtils.counter(EntityServiceImpl.class, "batch_with_noop_sysmetadata").inc();
+        return null;
+      }
     }
 
     // 4. Save the newValue as the latest version
-    if (!DataTemplateUtil.areEqual(databaseValue, writeItem.getRecordTemplate())) {
+    if (!DataTemplateUtil.areEqual(previousValue, writeItem.getRecordTemplate())) {
       log.debug(
           "Ingesting aspect with name {}, urn {}", writeItem.getAspectName(), writeItem.getUrn());
       String newValueStr = EntityApiUtils.toJsonAspect(writeItem.getRecordTemplate());

metadata-io/src/main/java/com/linkedin/metadata/entity/ebean/EbeanAspectDao.java

@@ -191,7 +191,7 @@ public long saveLatestAspect(
         newTime,
         newSystemMetadata,
         ASPECT_LATEST_VERSION,
-        oldAspectMetadata == null);
+        ASPECT_LATEST_VERSION.equals(nextVersion));
 
     return largestVersion;
   }

metadata-io/src/test/java/com/linkedin/metadata/entity/EbeanEntityServiceTest.java

@@ -538,7 +538,7 @@ public void testBatchPatchWithTrailingNoOp() throws Exception {
             opContext, DATASET_ENTITY_NAME, entityUrn, GLOBAL_TAGS_ASPECT_NAME);
     assertEquals(
         envelopedAspect.getSystemMetadata().getVersion(),
-        "3",
+        "2",
         "Expected version 3. 1 - Initial, + 1 add, 1 remove");
     assertEquals(
         new GlobalTags(envelopedAspect.getValue().data())
@@ -653,7 +653,7 @@ public void testBatchPatchAdd() throws Exception {
     EnvelopedAspect envelopedAspect =
         _entityServiceImpl.getLatestEnvelopedAspect(
             opContext, DATASET_ENTITY_NAME, entityUrn, GLOBAL_TAGS_ASPECT_NAME);
-    assertEquals(envelopedAspect.getSystemMetadata().getVersion(), "4", "Expected version 4");
+    assertEquals(envelopedAspect.getSystemMetadata().getVersion(), "3", "Expected version 4");
     assertEquals(
         new GlobalTags(envelopedAspect.getValue().data())
             .getTags().stream().map(TagAssociation::getTag).collect(Collectors.toSet()),
@@ -741,7 +741,7 @@ public void testBatchPatchAddDuplicate() throws Exception {
     EnvelopedAspect envelopedAspect =
         _entityServiceImpl.getLatestEnvelopedAspect(
             opContext, DATASET_ENTITY_NAME, entityUrn, GLOBAL_TAGS_ASPECT_NAME);
-    assertEquals(envelopedAspect.getSystemMetadata().getVersion(), "3", "Expected version 3");
+    assertEquals(envelopedAspect.getSystemMetadata().getVersion(), "2", "Expected version 2");
     assertEquals(
         new GlobalTags(envelopedAspect.getValue().data())
             .getTags().stream().map(TagAssociation::getTag).collect(Collectors.toSet()),

metadata-service/configuration/src/main/resources/application.yaml

@@ -159,7 +159,7 @@ ebean:
   autoCreateDdl: ${EBEAN_AUTOCREATE:false}
   postgresUseIamAuth: ${EBEAN_POSTGRES_USE_AWS_IAM_AUTH:false}
   locking:
-    enabled: ${EBEAN_LOCKING_ENABLED:true}
+    enabled: ${EBEAN_LOCKING_ENABLED:false}
     durationSeconds: ${EBEAN_LOCKING_DURATION_SECONDS:60}
     maximumLocks: ${EBEAN_LOCKING_MAXIMUM_LOCKS:20000}