Skip to content

Commit 326921e

Browse files
yan-3005yan-3005
authored
Remove Redundant updateMetadata from Workflow Set Action (#24907)
1 parent aa51aab commit 326921e

2 files changed

Lines changed: 165 additions & 56 deletions

File tree

‎openmetadata-service/src/main/java/org/openmetadata/service/util/EntityFieldUtils.java‎

Lines changed: 0 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -36,11 +36,6 @@
3636
import org.openmetadata.service.jdbi3.EntityRepository;
3737
import org.openmetadata.service.resources.tags.TagLabelUtil;
3838

39-
/**
40-
* Shared utility class for setting entity fields across different workflow components.
41-
* This provides a unified approach to field setting used by both ApprovalTaskWorkflow
42-
* and SetEntityAttributeImpl.
43-
*/
4439
@Slf4j
4540
public class EntityFieldUtils {
4641

@@ -77,61 +72,38 @@ public static void setEntityField(
7772
// Store original state for patch creation
7873
String originalJson = applyPatch ? JsonUtils.pojoToJson(entity) : null;
7974

80-
// Handle different field types
8175
switch (fieldName) {
8276
case "description":
8377
case "displayName":
8478
case "name":
85-
// Simple string fields
8679
setSimpleStringField(entity, fieldName, fieldValue);
8780
break;
88-
8981
case "tags":
90-
// Fetch Tag entities and append to existing tags
9182
appendTags(entity, fieldValue);
9283
break;
93-
9484
case "glossaryTerms":
95-
// Fetch GlossaryTerm entities and append to existing glossaryTerms
9685
appendGlossaryTerms(entity, fieldValue);
9786
break;
98-
9987
case "certification":
100-
// Set certification (replaces existing)
10188
setCertification(entity, fieldValue);
10289
break;
103-
10490
case "tier":
105-
// Set tier (replaces existing tier tag)
10691
setTier(entity, fieldValue);
10792
break;
108-
10993
case "owners":
110-
// Fetch User/Team entities and set as owners
11194
setOwners(entity, fieldValue);
11295
break;
113-
11496
case "reviewers":
115-
// Fetch User/Team entities and set as reviewers
11697
setReviewers(entity, fieldValue);
11798
break;
118-
11999
case "status":
120100
case "entityStatus":
121-
// Set entity status - handle different entity types appropriately
122101
setEntityStatus(entity, fieldValue);
123102
break;
124-
125103
default:
126-
// For other simple fields, try direct setting
127104
setSimpleStringField(entity, fieldName, fieldValue);
128105
break;
129106
}
130-
131-
// Update entity metadata
132-
updateEntityMetadata(entity, user);
133-
134-
// Create and apply patch if requested
135107
if (applyPatch) {
136108
String updatedJson = JsonUtils.pojoToJson(entity);
137109
JsonPatch patch = JsonUtils.getJsonPatch(originalJson, updatedJson);
@@ -155,9 +127,6 @@ public static void setEntityField(
155127
}
156128
}
157129

158-
/**
159-
* Sets a simple string field using reflection.
160-
*/
161130
public static void setSimpleStringField(
162131
EntityInterface entity, String fieldName, String fieldValue) {
163132
try {
@@ -283,8 +252,6 @@ public static void appendGlossaryTerms(EntityInterface entity, String termFQNs)
283252
if (termFQNs == null || termFQNs.isEmpty()) {
284253
return;
285254
}
286-
287-
// Get existing tags (glossary terms are stored as tags with source=GLOSSARY)
288255
List<TagLabel> existingTags = entity.getTags() != null ? entity.getTags() : new ArrayList<>();
289256
List<TagLabel> newTermsToAdd = new ArrayList<>();
290257
String[] fqns = termFQNs.contains(",") ? termFQNs.split(",") : new String[] {termFQNs};
@@ -326,10 +293,8 @@ public static void appendGlossaryTerms(EntityInterface entity, String termFQNs)
326293
// Check for mutual exclusivity
327294
try {
328295
TagLabelUtil.checkMutuallyExclusive(combinedTags);
329-
// If no exception, we can safely append
330296
entity.setTags(combinedTags);
331297
} catch (IllegalArgumentException e) {
332-
// Mutual exclusivity conflict detected - remove conflicting tags/terms and add new ones
333298
LOG.debug(
334299
"Mutual exclusivity conflict detected. Replacing conflicting glossary terms: {}",
335300
e.getMessage());
@@ -353,11 +318,7 @@ public static void appendGlossaryTerms(EntityInterface entity, String termFQNs)
353318
filteredTags.add(existingTag);
354319
}
355320
}
356-
357-
// Add the new terms
358321
filteredTags.addAll(newTermsToAdd);
359-
360-
// Final validation
361322
try {
362323
TagLabelUtil.checkMutuallyExclusive(filteredTags);
363324
entity.setTags(filteredTags);
@@ -526,26 +487,12 @@ public static void setReviewers(EntityInterface entity, String reviewerNames) {
526487
}
527488
}
528489

529-
/**
530-
* Update entity metadata (updatedBy, updatedAt) using reflection
531-
*/
532-
public static void updateEntityMetadata(EntityInterface entity, String user) {
533-
try {
534-
entity.setUpdatedBy(user);
535-
entity.setUpdatedAt(System.currentTimeMillis());
536-
} catch (Exception e) {
537-
LOG.debug("Failed to update entity metadata: {}", e.getMessage());
538-
}
539-
}
540-
541490
/**
542491
* Sets the status field on various entity types using appropriate enum values.
543492
* Handles both legacy 'status' field and new 'entityStatus' field.
544493
*/
545494
public static void setEntityStatus(EntityInterface entity, String statusValue) {
546495
try {
547-
// Try to parse as EntityStatus enum and use the EntityInterface method
548-
// This works for all entities that support entityStatus field
549496
try {
550497
EntityStatus status = EntityStatus.fromValue(statusValue);
551498
entity.setEntityStatus(status);
@@ -566,9 +513,6 @@ public static void setEntityStatus(EntityInterface entity, String statusValue) {
566513
"Entity type {} doesn't support entityStatus field, trying legacy status field",
567514
entity.getClass().getSimpleName());
568515
}
569-
570-
// Fallback: Try to set legacy "status" field for entities that don't support entityStatus
571-
// This maintains backward compatibility with older entities
572516
try {
573517
setSimpleStringField(entity, "status", statusValue);
574518
LOG.debug(

‎openmetadata-service/src/test/java/org/openmetadata/service/resources/governance/WorkflowDefinitionResourceTest.java‎

Lines changed: 165 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1807,6 +1807,171 @@ void test_EventBasedWorkflowForMultipleEntities(TestInfo test)
18071807

18081808
@Test
18091809
@Order(6)
1810+
void test_WorkflowFieldUpdateDoesNotCreateRedundantChangeEvents(TestInfo test) throws Exception {
1811+
LOG.info("Starting test to verify workflow field updates don't create redundant change events");
1812+
1813+
// Create a test table
1814+
CreateDatabaseService createService =
1815+
databaseServiceTest.createRequest(
1816+
"test_changeevent_service_" + test.getDisplayName().replaceAll("[^a-zA-Z0-9_]", ""));
1817+
DatabaseService service = databaseServiceTest.createEntity(createService, ADMIN_AUTH_HEADERS);
1818+
1819+
CreateDatabase createDatabase =
1820+
new CreateDatabase()
1821+
.withName(
1822+
"test_changeevent_db_" + test.getDisplayName().replaceAll("[^a-zA-Z0-9_]", ""))
1823+
.withService(service.getFullyQualifiedName());
1824+
Database database = databaseTest.createEntity(createDatabase, ADMIN_AUTH_HEADERS);
1825+
1826+
CreateDatabaseSchema createSchema =
1827+
new CreateDatabaseSchema()
1828+
.withName(
1829+
"test_changeevent_schema_" + test.getDisplayName().replaceAll("[^a-zA-Z0-9_]", ""))
1830+
.withDatabase(database.getFullyQualifiedName());
1831+
DatabaseSchema schema = schemaTest.createEntity(createSchema, ADMIN_AUTH_HEADERS);
1832+
1833+
CreateTable createTable =
1834+
new CreateTable()
1835+
.withName(
1836+
"test_changeevent_table_" + test.getDisplayName().replaceAll("[^a-zA-Z0-9_]", ""))
1837+
.withDatabaseSchema(schema.getFullyQualifiedName())
1838+
.withColumns(
1839+
List.of(
1840+
new Column().withName("id").withDataType(ColumnDataType.INT),
1841+
new Column().withName("name").withDataType(ColumnDataType.STRING)));
1842+
Table table = tableTest.createEntity(createTable, ADMIN_AUTH_HEADERS);
1843+
LOG.debug("Created test table: {}", table.getName());
1844+
1845+
// Record initial change event count
1846+
long initialOffset =
1847+
org.openmetadata.service.Entity.getCollectionDAO().changeEventDAO().getLatestOffset();
1848+
LOG.debug("Initial change event offset: {}", initialOffset);
1849+
1850+
// Create workflow that sets tags (this should create meaningful changes)
1851+
String workflowJson =
1852+
"""
1853+
{
1854+
"name": "testRedundantChangeEvents",
1855+
"displayName": "Test Redundant Change Events",
1856+
"description": "Test workflow to verify no redundant change events",
1857+
"trigger": {
1858+
"type": "periodicBatchEntity",
1859+
"config": {
1860+
"entityTypes": ["table"],
1861+
"schedule": {"scheduleTimeline": "None"},
1862+
"batchSize": 100,
1863+
"filters": {}
1864+
},
1865+
"output": ["relatedEntity", "updatedBy"]
1866+
},
1867+
"nodes": [
1868+
{"type": "startEvent", "subType": "startEvent", "name": "start", "displayName": "start"},
1869+
{
1870+
"type": "automatedTask",
1871+
"subType": "setEntityAttributeTask",
1872+
"name": "setTag",
1873+
"displayName": "Set Tag",
1874+
"config": {
1875+
"fieldName": "tags",
1876+
"fieldValue": "Tier.Tier1"
1877+
},
1878+
"input": ["relatedEntity", "updatedBy"],
1879+
"inputNamespaceMap": {"relatedEntity": "global", "updatedBy": "global"},
1880+
"output": []
1881+
},
1882+
{"type": "endEvent", "subType": "endEvent", "name": "end", "displayName": "end"}
1883+
],
1884+
"edges": [
1885+
{"from": "start", "to": "setTag"},
1886+
{"from": "setTag", "to": "end"}
1887+
],
1888+
"config": {"storeStageStatus": true}
1889+
}
1890+
""";
1891+
1892+
CreateWorkflowDefinition workflow =
1893+
JsonUtils.readValue(workflowJson, CreateWorkflowDefinition.class);
1894+
1895+
// Create and trigger workflow
1896+
Response response =
1897+
SecurityUtil.addHeaders(getResource("governance/workflowDefinitions"), ADMIN_AUTH_HEADERS)
1898+
.post(Entity.json(workflow));
1899+
assertTrue(
1900+
response.getStatus() == Response.Status.CREATED.getStatusCode()
1901+
|| response.getStatus() == Response.Status.OK.getStatusCode());
1902+
1903+
// Wait a moment for workflow setup
1904+
java.lang.Thread.sleep(2000);
1905+
1906+
// Trigger the workflow FIRST time
1907+
Response triggerResponse =
1908+
SecurityUtil.addHeaders(
1909+
getResource(
1910+
"governance/workflowDefinitions/name/testRedundantChangeEvents/trigger"),
1911+
ADMIN_AUTH_HEADERS)
1912+
.post(Entity.json("{}"));
1913+
assertEquals(Response.Status.OK.getStatusCode(), triggerResponse.getStatus());
1914+
1915+
// Wait for workflow to complete
1916+
java.lang.Thread.sleep(15000);
1917+
1918+
// Count change events after first workflow run
1919+
long firstRunOffset =
1920+
org.openmetadata.service.Entity.getCollectionDAO().changeEventDAO().getLatestOffset();
1921+
long firstRunEventCount = firstRunOffset - initialOffset;
1922+
1923+
// Verify the tag was actually added (meaningful change)
1924+
Table updatedTable = tableTest.getEntity(table.getId(), "tags", ADMIN_AUTH_HEADERS);
1925+
boolean hasTag =
1926+
updatedTable.getTags() != null
1927+
&& updatedTable.getTags().stream()
1928+
.anyMatch(tag -> "Tier.Tier1".equals(tag.getTagFQN()));
1929+
assertTrue(hasTag, "Table should have Tier.Tier1 tag after first workflow run");
1930+
1931+
LOG.info("First workflow run created {} change events", firstRunEventCount);
1932+
assertTrue(
1933+
firstRunEventCount > 0, "First workflow run should create at least one change event");
1934+
1935+
// Trigger the workflow SECOND time (should NOT create new events since no actual changes)
1936+
triggerResponse =
1937+
SecurityUtil.addHeaders(
1938+
getResource(
1939+
"governance/workflowDefinitions/name/testRedundantChangeEvents/trigger"),
1940+
ADMIN_AUTH_HEADERS)
1941+
.post(Entity.json("{}"));
1942+
assertEquals(Response.Status.OK.getStatusCode(), triggerResponse.getStatus());
1943+
1944+
// Wait for second workflow to complete
1945+
java.lang.Thread.sleep(15000);
1946+
1947+
// Count change events after second workflow run
1948+
long secondRunOffset =
1949+
org.openmetadata.service.Entity.getCollectionDAO().changeEventDAO().getLatestOffset();
1950+
long secondRunEventCount = secondRunOffset - firstRunOffset;
1951+
1952+
LOG.info("Second workflow run created {} change events", secondRunEventCount);
1953+
1954+
// CRITICAL ASSERTION: Second run should create NO new change events
1955+
// because the tag is already set and EntityFieldUtils should not generate
1956+
// redundant events due to updateEntityMetadata timestamp changes
1957+
assertEquals(
1958+
0,
1959+
secondRunEventCount,
1960+
"Second workflow run should NOT create change events when no actual field changes occur. "
1961+
+ "This verifies the fix for redundant updateEntityMetadata events.");
1962+
1963+
// Verify the tag is still there (no regression)
1964+
Table finalTable = tableTest.getEntity(table.getId(), "tags", ADMIN_AUTH_HEADERS);
1965+
boolean stillHasTag =
1966+
finalTable.getTags() != null
1967+
&& finalTable.getTags().stream().anyMatch(tag -> "Tier.Tier1".equals(tag.getTagFQN()));
1968+
assertTrue(stillHasTag, "Table should still have the tag after second workflow run");
1969+
1970+
LOG.info("✓ PASSED: Workflow field updates do not create redundant change events");
1971+
}
1972+
1973+
@Test
1974+
@Order(7)
18101975
void test_MultiEntityPeriodicQueryWithFilters(TestInfo test)
18111976
throws IOException, InterruptedException {
18121977
LOG.info("Starting test_MultiEntityPeriodicQueryWithFilters");

0 commit comments

Comments
 (0)