Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Skip generating task context reports for sub tasks (#17219) 

* Skip task context for sub tasks

* DRY a little + skip context for live report
This commit is contained in:
Arun Ramani 2024-10-02 15:32:50 +02:00 committed by GitHub
parent 65277b17a9
commit e5d027ee1c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 48 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
View File

@ -930,7 +930,24 @@ public abstract class AbstractBatchIndexTask extends AbstractTask
null,
null
)
),
)
);
}
/**
* Builds a map with the following keys and values:
* {@link IngestionStatsAndErrorsTaskReport#REPORT_KEY} : {@link IngestionStatsAndErrorsTaskReport}.
* {@link TaskContextReport#REPORT_KEY} : {@link TaskContextReport}.
*/
protected TaskReport.ReportMap buildIngestionStatsAndContextReport(
IngestionState ingestionState,
String errorMessage,
Long segmentsRead,
Long segmentsPublished
)
{
return TaskReport.buildTaskReports(
buildIngestionStatsTaskReport(ingestionState, errorMessage, segmentsRead, segmentsPublished),
new TaskContextReport(getId(), getContext())
);
}
@ -947,21 +964,33 @@ public abstract class AbstractBatchIndexTask extends AbstractTask
)
{
return TaskReport.buildTaskReports(
new IngestionStatsAndErrorsTaskReport(
getId(),
new IngestionStatsAndErrors(
ingestionState,
getTaskCompletionUnparseableEvents(),
getTaskCompletionRowStats(),
errorMessage,
segmentAvailabilityConfirmationCompleted,
segmentAvailabilityWaitTimeMs,
Collections.emptyMap(),
segmentsRead,
segmentsPublished
)
),
new TaskContextReport(getId(), getContext())
buildIngestionStatsTaskReport(ingestionState, errorMessage, segmentsRead, segmentsPublished)
);
}
/**
* Helper method to create IngestionStatsAndErrorsTaskReport.
*/
private IngestionStatsAndErrorsTaskReport buildIngestionStatsTaskReport(
IngestionState ingestionState,
String errorMessage,
Long segmentsRead,
Long segmentsPublished
)
{
return new IngestionStatsAndErrorsTaskReport(
getId(),
new IngestionStatsAndErrors(
ingestionState,
getTaskCompletionUnparseableEvents(),
getTaskCompletionRowStats(),
errorMessage,
segmentAvailabilityConfirmationCompleted,
segmentAvailabilityWaitTimeMs,
Collections.emptyMap(),
segmentsRead,
segmentsPublished
)
);
}

2
indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java
View File

@ -677,7 +677,7 @@ public class HadoopIndexTask extends HadoopTask implements ChatHandler
private TaskReport.ReportMap getTaskCompletionReports()
{
return buildIngestionStatsReport(ingestionState, errorMsg, null, null);
return buildIngestionStatsAndContextReport(ingestionState, errorMsg, null, null);
}
@Override

2
indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
View File

@ -537,7 +537,7 @@ public class IndexTask extends AbstractBatchIndexTask implements ChatHandler, Pe
private void updateAndWriteCompletionReports(TaskToolbox toolbox, Long segmentsRead, Long segmentsPublished)
{
completionReports = buildIngestionStatsReport(ingestionState, errorMsg, segmentsRead, segmentsPublished);
completionReports = buildIngestionStatsAndContextReport(ingestionState, errorMsg, segmentsRead, segmentsPublished);
if (isStandAloneTask) {
toolbox.getTaskReportFileWriter().write(getId(), completionReports);
}

2
indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
View File

@ -1251,7 +1251,7 @@ public class ParallelIndexSupervisorTask extends AbstractBatchIndexTask
*/
private TaskReport.ReportMap getTaskCompletionReports(TaskStatus taskStatus)
{
return buildIngestionStatsReport(
return buildIngestionStatsAndContextReport(
IngestionState.COMPLETED,
taskStatus.getErrorMsg(),
segmentsRead,