Merge pull request #1726 from AnuGayan/up_master_imp

Improve incremental aggregation error handling
siddhi-io · May 31, 2021 · 97e45e5 · 97e45e5
2 parents 1d33f6d + e4409be
commit 97e45e5
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 10 deletions.
diff --git a/modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalExecutor.java b/modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalExecutor.java
@@ -215,17 +215,19 @@ private void dispatchEvent(long startTimeOfNewAggregates, BaseIncrementalValueSt
                 LOG.debug("Event dispatched by " + this.duration + " incremental executor: " + eventChunk.toString());
             }
             if (isProcessingExecutor) {
-                try {
-                    executorService.execute(() -> {
+                executorService.execute(() -> {
+                    try {
                         table.addEvents(tableEventChunk, streamEventMap.size());
+                    } catch (Throwable t) {
+                        LOG.error("Exception occurred at siddhi app '" + this.siddhiAppName +
+                                "' when performing table writes of aggregation '" + this.aggregatorName +
+                                "' for duration '" + this.duration + "'. This should be investigated as this " +
+                                "can cause accuracy loss.", t);
+                    } finally {
                         isProcessFinished.set(true);
-                    });
-                } catch (Throwable t) {
-                    LOG.error("Exception occurred at siddhi app '" + this.siddhiAppName +
-                            "' when performing table writes of aggregation '" + this.aggregatorName +
-                            "' for duration '" + this.duration + "'. This should be investigated as this " +
-                            "can cause accuracy loss.", t);
-                }
+                    }
+                });
+
             }
             if (waitUntillprocessFinish) {
                 try {

diff --git a/modules/siddhi-core/src/main/java/io/siddhi/core/util/parser/AggregationParser.java b/modules/siddhi-core/src/main/java/io/siddhi/core/util/parser/AggregationParser.java
@@ -1255,7 +1255,6 @@ private static String generateDatabaseQuery(List<ExpressionExecutor> expressionE
                             .equals(AGG_START_TIMESTAMP_COL)) {
                         outerSelectColumnJoiner.add(" ? " + SQL_AS + variableExpressionExecutor.getAttribute().getName());
                     } else if (!variableExpressionExecutor.getAttribute().getName().equals(AGG_EXTERNAL_TIMESTAMP_COL)) {
-                        subSelectT2ColumnJoiner.add(variableExpressionExecutor.getAttribute().getName());
                         if (groupByColumnNames.contains(variableExpressionExecutor.getAttribute().getName())) {
                             subSelectT2ColumnJoiner.add(INNER_SELECT_QUERY_REF_T3 + "." +
                                     variableExpressionExecutor.getAttribute().getName() + SQL_AS +