Fix Expression Parsing (#14753)

open-metadata · Jan 17, 2024 · 5bf3201 · 5bf3201
1 parent 1ae94bb
commit 5bf3201
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 30 deletions.
diff --git a/...rc/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java b/...rc/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java
@@ -273,37 +273,42 @@ public List<ChangeEvent> pollEvents(long offset, long batchSize) {
   @Override
   public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
     // Must Have , Before Execute the Init, Quartz Requires a Non-Arg Constructor
-    this.init(jobExecutionContext);
-
-    // Poll Events from Change Event Table
-    List<ChangeEvent> batch = pollEvents(offset, 100);
-    int batchSize = batch.size();
-
-    // Retry Failed Events
-    Set<FailedEvent> failedEventsList =
-        JsonUtils.convertValue(
-            jobDetail.getJobDataMap().get(FAILED_EVENT_EXTENSION), new TypeReference<>() {});
-    if (failedEventsList != null) {
-      List<ChangeEvent> failedChangeEvents =
-          failedEventsList.stream()
-              .filter(failedEvent -> failedEvent.getRetriesLeft() > 0)
-              .map(FailedEvent::getChangeEvent)
-              .toList();
-      batch.addAll(failedChangeEvents);
-    }
+    try {
 
-    if (!batch.isEmpty()) {
-      // Publish Events
-      alertMetrics.withTotalEvents(alertMetrics.getTotalEvents() + batch.size());
-      publishEvents(batch);
+      this.init(jobExecutionContext);
 
-      // Commit the Offset
-      offset += batchSize;
-      commit(jobExecutionContext);
-    }
+      // Poll Events from Change Event Table
+      List<ChangeEvent> batch = pollEvents(offset, 100);
+      int batchSize = batch.size();
+
+      // Retry Failed Events
+      Set<FailedEvent> failedEventsList =
+          JsonUtils.convertValue(
+              jobDetail.getJobDataMap().get(FAILED_EVENT_EXTENSION), new TypeReference<>() {});
+      if (failedEventsList != null) {
+        List<ChangeEvent> failedChangeEvents =
+            failedEventsList.stream()
+                .filter(failedEvent -> failedEvent.getRetriesLeft() > 0)
+                .map(FailedEvent::getChangeEvent)
+                .toList();
+        batch.addAll(failedChangeEvents);
+      }
+
+      if (!batch.isEmpty()) {
+        // Publish Events
+        alertMetrics.withTotalEvents(alertMetrics.getTotalEvents() + batch.size());
+        publishEvents(batch);
 
-    // Call stop to close the client
-    this.stop();
+        // Commit the Offset
+        offset += batchSize;
+        commit(jobExecutionContext);
+      }
+    } catch (Exception e) {
+      LOG.error("Error in executing the Job : {} ", e.getMessage());
+    } finally {
+      // Call stop to close the client
+      this.stop();
+    }
   }
 
   public EventSubscription getEventSubscription() {

diff --git a/openmetadata-ui/src/main/resources/ui/src/pages/AddAlertPage/AddAlertPage.tsx b/openmetadata-ui/src/main/resources/ui/src/pages/AddAlertPage/AddAlertPage.tsx
@@ -183,10 +183,10 @@ const AddAlertPage = () => {
 
     const requestFilteringRules = filteringRules?.rules?.map((curr) => ({
       ...curr,
-      condition: `${curr.name}(${map(
+      condition: `${curr.name}({${map(
         curr.condition,
         (v: string) => `'${v}'`
-      )?.join(', ')})`,
+      )?.join(', ')}})`,
     }));
 
     try {