Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,12 @@

package org.apache.druid.indexing.common.task;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.Files;
import org.apache.druid.indexer.IngestionState;
import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReport;
import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReportData;
Expand All @@ -34,6 +37,7 @@
import org.junit.rules.TemporaryFolder;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class TaskReportSerdeTest
Expand All @@ -47,6 +51,7 @@ public TaskReportSerdeTest()
{
TestUtils testUtils = new TestUtils();
jsonMapper = testUtils.getTestObjectMapper();
jsonMapper.registerSubtypes(ExceptionalTaskReport.class);
}

@Test
Expand Down Expand Up @@ -87,4 +92,47 @@ public void testSerde() throws Exception
);
Assert.assertEquals(reportMap1, reportMap2);
}

@Test
public void testExceptionWhileWritingReport() throws Exception
{
final File reportFile = temporaryFolder.newFile();
final SingleFileTaskReportFileWriter writer = new SingleFileTaskReportFileWriter(reportFile);
writer.setObjectMapper(jsonMapper);
writer.write("theTask", ImmutableMap.of("report", new ExceptionalTaskReport()));

// Read the file, ensure it's incomplete and not valid JSON. This allows callers to determine the report was
// not complete when written.
Assert.assertEquals(
"{\"report\":{\"type\":\"exceptional\"",
Files.asCharSource(reportFile, StandardCharsets.UTF_8).read()
);
}

/**
* Task report that throws an exception while being serialized.
*/
@JsonTypeName("exceptional")
private static class ExceptionalTaskReport implements TaskReport
{
@Override
@JsonProperty
public String getTaskId()
{
throw new UnsupportedOperationException("cannot serialize task ID");
}

@Override
public String getReportKey()
{
return "report";
}

@Override
@JsonProperty
public Object getPayload()
{
throw new UnsupportedOperationException("cannot serialize payload");
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.apache.druid.jackson;

import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JavaType;
Expand All @@ -35,7 +36,6 @@
import org.apache.druid.java.util.common.StringUtils;

import javax.annotation.Nullable;

import java.io.IOException;

/**
Expand Down Expand Up @@ -81,6 +81,10 @@ public DefaultObjectMapper(JsonFactory factory, @Nullable String serviceName)
configure(SerializationFeature.INDENT_OUTPUT, false);
configure(SerializationFeature.FLUSH_AFTER_WRITE_VALUE, false);

// Disable automatic JSON termination, so readers can detect truncated responses when a JsonGenerator is
// closed after an exception is thrown while writing.
configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);

addHandler(new DefaultDeserializationProblemHandler(serviceName));
}

Expand Down
3 changes: 0 additions & 3 deletions sql/src/main/java/org/apache/druid/sql/http/ArrayWriter.java
Original file line number Diff line number Diff line change
Expand Up @@ -43,9 +43,6 @@ public ArrayWriter(final OutputStream outputStream, final ObjectMapper jsonMappe
this.serializers = jsonMapper.getSerializerProviderInstance();
this.jsonGenerator = jsonMapper.getFactory().createGenerator(outputStream);
this.outputStream = outputStream;

// Disable automatic JSON termination, so clients can detect truncated responses.
jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);
}

@Override
Expand Down
3 changes: 0 additions & 3 deletions sql/src/main/java/org/apache/druid/sql/http/ObjectWriter.java
Original file line number Diff line number Diff line change
Expand Up @@ -46,9 +46,6 @@ public ObjectWriter(final OutputStream outputStream, final ObjectMapper jsonMapp
this.serializers = jsonMapper.getSerializerProviderInstance();
this.jsonGenerator = jsonMapper.getFactory().createGenerator(outputStream);
this.outputStream = outputStream;

// Disable automatic JSON termination, so clients can detect truncated responses.
jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);
}

@Override
Expand Down