This is an automated email from the ASF dual-hosted git repository.
adarshsanjeev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new c5127ba92fe Get reports file from file writer (#18379)
c5127ba92fe is described below
commit c5127ba92fe32d031fa25a10ef302e2d53e871c8
Author: Adarsh Sanjeev <[email protected]>
AuthorDate: Tue Aug 12 10:03:41 2025 +0530
Get reports file from file writer (#18379)
Switch to using the reports file from the reports writer when pushing the
reports. This fixes a bug where the AbstractTask could be unable to find the
report.
---
.../druid/indexing/common/MultipleFileTaskReportFileWriter.java | 6 ++++++
.../java/org/apache/druid/indexing/common/task/AbstractTask.java | 2 +-
.../org/apache/druid/indexing/overlord/duty/KillTaskToolbox.java | 9 +++++++++
.../indexing/common/MultipleFileTaskReportFileWriterTest.java | 2 ++
.../indexing/common/SingleFileTaskReportFileWriterTest.java | 2 ++
.../org/apache/druid/indexing/common/task/AbstractTaskTest.java | 4 ++++
.../druid/indexing/common/task/NoopTestTaskReportFileWriter.java | 7 +++++++
.../druid/indexer/report/SingleFileTaskReportFileWriter.java | 6 ++++++
.../org/apache/druid/indexer/report/TaskReportFileWriter.java | 8 ++++++++
9 files changed, 45 insertions(+), 1 deletion(-)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/MultipleFileTaskReportFileWriter.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/MultipleFileTaskReportFileWriter.java
index 865a8593a7d..edf8e028b91 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/MultipleFileTaskReportFileWriter.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/MultipleFileTaskReportFileWriter.java
@@ -75,6 +75,12 @@ public class MultipleFileTaskReportFileWriter implements
TaskReportFileWriter
this.objectMapper = objectMapper;
}
+ @Override
+ public File getReportsFile(String taskId)
+ {
+ return taskReportFiles.get(taskId);
+ }
+
public void add(String taskId, File reportsFile)
{
taskReportFiles.put(taskId, reportsFile);
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractTask.java
index a3331767ba9..aa205900291 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractTask.java
@@ -153,7 +153,7 @@ public abstract class AbstractTask implements Task
FileUtils.mkdirp(taskDir);
File attemptDir = Paths.get(taskDir.getAbsolutePath(), "attempt",
toolbox.getAttemptId()).toFile();
FileUtils.mkdirp(attemptDir);
- reportsFile = new File(attemptDir, "report.json");
+ reportsFile = toolbox.getTaskReportFileWriter().getReportsFile(getId());
statusFile = new File(attemptDir, "status.json");
InetAddress hostName = InetAddress.getLocalHost();
DruidNode node = toolbox.getTaskExecutorNode();
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/KillTaskToolbox.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/KillTaskToolbox.java
index dd2afe6017d..5451b3711e9 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/KillTaskToolbox.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/KillTaskToolbox.java
@@ -33,6 +33,8 @@ import org.apache.druid.segment.column.ColumnConfig;
import org.apache.druid.segment.loading.DataSegmentKiller;
import org.apache.druid.segment.writeout.TmpFileSegmentWriteOutMediumFactory;
+import javax.annotation.Nullable;
+import java.io.File;
import java.io.OutputStream;
/**
@@ -82,6 +84,13 @@ public class KillTaskToolbox
// Do nothing
}
+ @Override
+ @Nullable
+ public File getReportsFile(String taskId)
+ {
+ return null;
+ }
+
@Override
public void write(String taskId, TaskReport.ReportMap reports)
{
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/MultipleFileTaskReportFileWriterTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/MultipleFileTaskReportFileWriterTest.java
index 2e51973ec24..3ded86c1c43 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/MultipleFileTaskReportFileWriterTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/MultipleFileTaskReportFileWriterTest.java
@@ -60,5 +60,7 @@ public class MultipleFileTaskReportFileWriterTest
reportsMap,
mapper.readValue(Files.readAllBytes(file.toPath()), new
TypeReference<Map<String, TaskReport>>() {})
);
+
+ Assert.assertEquals(file.getAbsolutePath(),
writer.getReportsFile(TASK_ID).getAbsolutePath());
}
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/SingleFileTaskReportFileWriterTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/SingleFileTaskReportFileWriterTest.java
index 1381a7483cb..057ef7bfa05 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/SingleFileTaskReportFileWriterTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/SingleFileTaskReportFileWriterTest.java
@@ -57,5 +57,7 @@ public class SingleFileTaskReportFileWriterTest
reportsMap,
mapper.readValue(Files.readAllBytes(file.toPath()), new
TypeReference<Map<String, TaskReport>>() {})
);
+
+ Assert.assertEquals(file.getAbsolutePath(),
writer.getReportsFile(TASK_ID).getAbsolutePath());
}
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/AbstractTaskTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/AbstractTaskTest.java
index ba210fee228..987bf988db0 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/AbstractTaskTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/AbstractTaskTest.java
@@ -22,6 +22,7 @@ package org.apache.druid.indexing.common.task;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.io.FileUtils;
import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexer.report.SingleFileTaskReportFileWriter;
import org.apache.druid.indexing.common.TaskToolbox;
import org.apache.druid.indexing.common.TestUtils;
import org.apache.druid.indexing.common.actions.TaskActionClient;
@@ -88,6 +89,7 @@ public class AbstractTaskTest
TaskActionClient taskActionClient = mock(TaskActionClient.class);
when(taskActionClient.submit(any())).thenReturn(TaskConfig.class);
when(toolbox.getTaskActionClient()).thenReturn(taskActionClient);
+ when(toolbox.getTaskReportFileWriter()).thenReturn(new
SingleFileTaskReportFileWriter(temporaryFolder.newFile()));
AbstractTask task = new NoopTask("myID", null, null, 1, 0, null)
@@ -181,6 +183,7 @@ public class AbstractTaskTest
TaskActionClient taskActionClient = mock(TaskActionClient.class);
when(taskActionClient.submit(any())).thenReturn(TaskConfig.class);
when(toolbox.getTaskActionClient()).thenReturn(taskActionClient);
+ when(toolbox.getTaskReportFileWriter()).thenReturn(new
SingleFileTaskReportFileWriter(temporaryFolder.newFile()));
TaskStatus taskStatus = TaskStatus.failure("myId", "failed");
AbstractTask task = new NoopTask("myID", null, null, 1, 0, null)
@@ -201,6 +204,7 @@ public class AbstractTaskTest
{
TaskToolbox toolbox = mock(TaskToolbox.class);
when(toolbox.getAttemptId()).thenReturn("1");
+ when(toolbox.getTaskReportFileWriter()).thenReturn(new
SingleFileTaskReportFileWriter(temporaryFolder.newFile()));
DruidNode node = new DruidNode("foo", "foo", false, 1, 2, true, true);
when(toolbox.getTaskExecutorNode()).thenReturn(node);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/NoopTestTaskReportFileWriter.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/NoopTestTaskReportFileWriter.java
index ad175faeb49..ff52e966657 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/NoopTestTaskReportFileWriter.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/NoopTestTaskReportFileWriter.java
@@ -24,6 +24,7 @@ import org.apache.druid.indexer.report.TaskReport;
import org.apache.druid.indexer.report.TaskReportFileWriter;
import java.io.ByteArrayOutputStream;
+import java.io.File;
import java.io.OutputStream;
public class NoopTestTaskReportFileWriter implements TaskReportFileWriter
@@ -45,4 +46,10 @@ public class NoopTestTaskReportFileWriter implements
TaskReportFileWriter
{
}
+
+ @Override
+ public File getReportsFile(String taskId)
+ {
+ return null;
+ }
}
diff --git
a/processing/src/main/java/org/apache/druid/indexer/report/SingleFileTaskReportFileWriter.java
b/processing/src/main/java/org/apache/druid/indexer/report/SingleFileTaskReportFileWriter.java
index 9012f4e83a1..efa77fc2e30 100644
---
a/processing/src/main/java/org/apache/druid/indexer/report/SingleFileTaskReportFileWriter.java
+++
b/processing/src/main/java/org/apache/druid/indexer/report/SingleFileTaskReportFileWriter.java
@@ -68,6 +68,12 @@ public class SingleFileTaskReportFileWriter implements
TaskReportFileWriter
this.objectMapper = objectMapper;
}
+ @Override
+ public File getReportsFile(String taskId)
+ {
+ return reportsFile;
+ }
+
public static void writeReportToStream(
final ObjectMapper objectMapper,
final OutputStream outputStream,
diff --git
a/processing/src/main/java/org/apache/druid/indexer/report/TaskReportFileWriter.java
b/processing/src/main/java/org/apache/druid/indexer/report/TaskReportFileWriter.java
index 0cdd0249366..d4f1c6fd959 100644
---
a/processing/src/main/java/org/apache/druid/indexer/report/TaskReportFileWriter.java
+++
b/processing/src/main/java/org/apache/druid/indexer/report/TaskReportFileWriter.java
@@ -21,6 +21,8 @@ package org.apache.druid.indexer.report;
import com.fasterxml.jackson.databind.ObjectMapper;
+import javax.annotation.Nullable;
+import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
@@ -31,4 +33,10 @@ public interface TaskReportFileWriter
OutputStream openReportOutputStream(String taskId) throws IOException;
void setObjectMapper(ObjectMapper objectMapper);
+
+ /**
+ * Returns the reports file for the given taskId. Returns null if writer
does not write reports to any file.
+ */
+ @Nullable
+ File getReportsFile(String taskId);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
