address comments

apache · Jan 10, 2025 · 6dba948 · 6dba948
1 parent b5e6879
commit 6dba948
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 18 deletions.
diff --git a/...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java b/...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java
@@ -555,7 +555,7 @@ private void emitSummaryMetrics(final MSQTaskReportPayload msqTaskReportPayload,
                 }).sum()).sum()
         : 0;
 
-    log.info("Total processed bytes: %d, query: %s", totalProcessedBytes, querySpec.getQuery());
+    log.debug("Processed bytes[%d] for query[%s].", totalProcessedBytes, querySpec.getQuery());
     context.emitMetric("ingest/processed/bytes", totalProcessedBytes);
   }
 

diff --git a/...ava/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java b/...ava/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
@@ -1634,12 +1634,7 @@ private Pair<Map<String, Object>, Map<String, Object>> doGetRowStatsAndUnparseab
     buildSegmentsRowStats.addRowIngestionMetersTotals(rowStatsForRunningTasks);
 
     // Emit the processed bytes metric
-    try {
-      emitMetric(toolbox.getEmitter(), "ingest/processed/bytes", rowStatsForRunningTasks.getProcessedBytes());
-    }
-    catch (Exception e) {
-      LOG.warn(e, "Unable to emit processed bytes metric");
-    }
+    emitMetric(toolbox.getEmitter(), "ingest/processed/bytes", rowStatsForRunningTasks.getProcessedBytes());
 
     return createStatsAndErrorsReport(buildSegmentsRowStats.getTotals(), unparseableEvents);
   }

diff --git a/...src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java b/...src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
@@ -664,17 +664,10 @@ public void run()
             }
 
             // Emit the processed bytes metric
-            try {
-              toolbox.getEmitter().emit(
-                  ServiceMetricEvent.builder()
-                      .setDimension("taskId", task.getId())
-                      .setDimension("dataSource", task.getDataSource())
-                      .setMetric("ingest/processed/bytes", bytesProcessed)
-              );
-            }
-            catch (Exception e) {
-              log.warn(e, "Unable to emit processed bytes metric");
-            }
+            IndexTaskUtils.setTaskDimensions(new ServiceMetricEvent.Builder(), task);
+            toolbox.getEmitter().emit(
+                ServiceMetricEvent.builder().setMetric("ingest/processed/bytes", bytesProcessed)
+            );
 
             if (shouldProcess) {
               final List<InputRow> rows = parser.parse(record.getData(), isEndOfShard(record.getSequenceNumber()));