This is an automated email from the ASF dual-hosted git repository.
KKcorps pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new c5b8107fa6a Return the cause chain when an /ingestFromFile record
fails (#19716)
c5b8107fa6a is described below
commit c5b8107fa6a0e0bcf7f1b401cc2aa827a7e32e74
Author: Anshul Singh <[email protected]>
AuthorDate: Mon Oct 26 12:18:23 2026 +0530
Return the cause chain when an /ingestFromFile record fails (#19716)
* Return the cause chain when an /ingestFromFile record fails
An upload with one bad value answered "Caught exception while reading data":
the controller logged the column and the value, but the response kept only
the outermost message.
- ControllerApplicationException.withCausesInResponse keeps the cause for
the
response. Every constructor behaves as before.
- WebApplicationExceptionMapper appends ExceptionUtils.appendCauses for
those
exceptions only: each cause after " -> ", skipping a message the text
already contains, with the appended part cut in the middle to 2,048
characters so the root cause stays at the end. Each cause is cut to that
budget as the chain is walked, so the work stays bounded however long a
record value in a message is.
- /ingestFromFile opts in when the chain holds the new
RecordProcessingException, which segment creation throws for a record it
cannot read or transform. Setup failures, which can involve files on the
controller, keep their message only.
* Take the response cause chain from the RecordProcessingException
Wrappers between the caught exception and the record failure come from code
that is not reading the record, so they stay out of the response. The full
exception is still logged by FileIngestionHelper. Call commons-lang3's
ExceptionUtils by its full name so it is not confused with Pinot's.
* Retrigger CI: the Kafka 4.0 test container failed to start
---
.../apache/pinot/common/utils/ExceptionUtils.java | 45 +++++++
.../pinot/common/utils/ExceptionUtilsTest.java | 129 +++++++++++++++++++
.../exception/ControllerApplicationException.java | 23 +++-
.../resources/PinotIngestionRestletResource.java | 27 +++-
.../resources/WebApplicationExceptionMapper.java | 19 ++-
...PinotIngestionRestletResourceStatelessTest.java | 86 ++++++++++++-
.../ControllerApplicationExceptionTest.java | 36 +++++-
.../PinotIngestionRestletResourceTest.java | 140 ++++++++++++++++++++
.../WebApplicationExceptionMapperTest.java | 98 ++++++++++++++
.../segment/creator/RecordProcessingException.java | 29 +++++
.../RecordReaderSegmentCreationDataSource.java | 2 +-
.../impl/SegmentIndexCreationDriverImpl.java | 3 +-
.../creator/RecordProcessingExceptionTest.java | 143 +++++++++++++++++++++
13 files changed, 768 insertions(+), 12 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ExceptionUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ExceptionUtils.java
index 37da64b14ee..17a437e6aea 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ExceptionUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ExceptionUtils.java
@@ -18,12 +18,17 @@
*/
package org.apache.pinot.common.utils;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Set;
import java.util.function.Function;
import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
public class ExceptionUtils {
+ private static final String CAUSE_SEPARATOR = " -> ";
+ private static final String ELLIPSIS = "...";
private ExceptionUtils() {
}
@@ -59,6 +64,46 @@ public class ExceptionUtils {
return sb.toString();
}
+ /// Appends the messages of `cause` and its causes to `message`, each after
`" -> "`, skipping any message the text
+ /// already contains. Each cause, then the appended part, is abbreviated in
the middle to `maxLength` characters,
+ /// which keeps the root cause at the end and the work bounded however long
a cause's message is.
+ @Nullable
+ public static String appendCauses(@Nullable String message, @Nullable
Throwable cause, int maxLength) {
+ String text = message != null ? message : "";
+ StringBuilder causes = new StringBuilder();
+ Set<Throwable> visited = Collections.newSetFromMap(new
IdentityHashMap<>());
+ for (Throwable t = cause; t != null && visited.add(t); t = t.getCause()) {
+ String causeMessage = t.getMessage();
+ if (StringUtils.isBlank(causeMessage) || isGeneratedFrom(t.getCause(),
causeMessage)
+ || text.contains(causeMessage)) {
+ continue;
+ }
+ String shownMessage = StringUtils.abbreviateMiddle(causeMessage,
ELLIPSIS, maxLength);
+ if (causes.indexOf(shownMessage) < 0) {
+ causes.append(CAUSE_SEPARATOR).append(shownMessage);
+ }
+ }
+ if (causes.length() == 0) {
+ return message;
+ }
+ String chain = text.isEmpty() ? causes.substring(CAUSE_SEPARATOR.length())
: causes.toString();
+ return text + StringUtils.abbreviateMiddle(chain, ELLIPSIS, maxLength);
+ }
+
+ /// Whether `message` is the one a wrapper such as `new
RuntimeException(cause)` takes from its cause, i.e. equals
+ /// `cause.toString()`, checked without building that string.
+ private static boolean isGeneratedFrom(@Nullable Throwable cause, String
message) {
+ if (cause == null) {
+ return false;
+ }
+ String className = cause.getClass().getName();
+ String causeMessage = cause.getLocalizedMessage();
+ if (causeMessage == null) {
+ return message.equals(className);
+ }
+ return message.length() == className.length() + 2 + causeMessage.length()
&& message.startsWith(className)
+ && message.startsWith(": ", className.length()) &&
message.endsWith(causeMessage);
+ }
public static String getStackTrace(Throwable e) {
return getStackTrace(e, Integer.MAX_VALUE);
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/ExceptionUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ExceptionUtilsTest.java
new file mode 100644
index 00000000000..df80f54f0c4
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ExceptionUtilsTest.java
@@ -0,0 +1,129 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.utils;
+
+import java.io.IOException;
+import org.apache.commons.lang3.StringUtils;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests for [ExceptionUtils#appendCauses(String, Throwable, int)].
+public class ExceptionUtilsTest {
+ private static final int MAX_LENGTH = 1024;
+
+ @Test
+ public void testAppendCausesReturnsTheMessageWhenThereIsNoCause() {
+ String message = "Table: foo_OFFLINE not found";
+
+ assertSame(append(message, null), message);
+ assertNull(append(null, null));
+ }
+
+ @Test
+ public void testAppendCausesListsCausesOutermostFirst() {
+ Exception cause = new IllegalStateException("middle", new
IOException("inner"));
+
+ assertEquals(append("outer", cause), "outer -> middle -> inner");
+ }
+
+ /// The `/ingestFromFile` response for a record that cannot be read: the
wrapper message already ends with the first
+ /// cause, so only the two below it are added.
+ @Test
+ public void testAppendCausesKeepsTheRootCauseOfASegmentCreationFailure() {
+ Exception rootCause = new IllegalArgumentException(
+ "Cannot read single-value from Object[]: [before, after] for column:
New_Value");
+ Exception transformFailure =
+ new RuntimeException("Caught exception while transforming data type
for column: New_Value", rootCause);
+ Exception readFailure = new RuntimeException("Caught exception while
reading data", transformFailure);
+ String message = "Caught exception when ingesting file into table:
foo_OFFLINE. " + readFailure.getMessage();
+
+ assertEquals(append(message, readFailure), message
+ + " -> Caught exception while transforming data type for column:
New_Value"
+ + " -> Cannot read single-value from Object[]: [before, after] for
column: New_Value");
+ }
+
+ @Test
+ public void testAppendCausesWithoutMessageStartsWithTheFirstCause() {
+ Exception cause = new IllegalStateException("middle", new
IOException("inner"));
+
+ assertEquals(append(null, cause), "middle -> inner");
+ assertEquals(append("", cause), "middle -> inner");
+ }
+
+ @Test
+ public void testAppendCausesSkipsMessagesTheTextAlreadyContains() {
+ Exception cause = new IllegalStateException("Segment push failed", new
IllegalStateException(
+ "Failed to read schema: Unexpected character at line 3",
+ new IllegalStateException("Unexpected character at line 3", new
IOException("Connection reset"))));
+
+ assertEquals(append("Segment push failed", cause),
+ "Segment push failed -> Failed to read schema: Unexpected character at
line 3 -> Connection reset");
+ }
+
+ @Test
+ public void
testAppendCausesSkipsBlankMessagesAndMessagesGeneratedFromTheCause() {
+ Exception cause = new RuntimeException(
+ new RuntimeException(new IllegalStateException(" ", new
IllegalArgumentException("Invalid column: foo"))));
+
+ assertEquals(append("Failed to build segment", cause), "Failed to build
segment -> Invalid column: foo");
+ assertEquals(append("Failed to build segment", new RuntimeException(new
NullPointerException())),
+ "Failed to build segment");
+ }
+
+ @Test
+ public void
testAppendCausesKeepsAMessageThatOnlyLooksGeneratedFromTheCause() {
+ Exception cause = new
IllegalStateException("java.lang.IllegalArgumentException: something else",
+ new IllegalArgumentException("root"));
+
+ assertEquals(append("top", cause), "top ->
java.lang.IllegalArgumentException: something else -> root");
+ }
+
+ @Test
+ public void
testAppendCausesAbbreviatesTheCausesInTheMiddleButNotTheMessage() {
+ String message = StringUtils.repeat('m', 200);
+ String rootCause = "Cannot read single-value from Object[]: [a, b] for
column: v";
+ Exception cause = new IllegalStateException(StringUtils.repeat('x', 5000),
new IllegalArgumentException(rootCause));
+
+ String result = ExceptionUtils.appendCauses(message, cause, 200);
+
+ assertEquals(result.length(), message.length() + 200);
+ assertTrue(result.startsWith(message + " -> xxx"), result);
+ assertTrue(result.contains("x...x"), result);
+ assertTrue(result.endsWith("x -> " + rootCause), result);
+ }
+
+ @Test(timeOut = 10_000)
+ public void testAppendCausesTerminatesOnACauseCycle() {
+ Exception first = new RuntimeException("first");
+ Exception second = new IllegalStateException("second");
+ first.initCause(second);
+ second.initCause(first);
+
+ assertEquals(append("top", first), "top -> first -> second");
+ }
+
+ private static String append(String message, Throwable cause) {
+ return ExceptionUtils.appendCauses(message, cause, MAX_LENGTH);
+ }
+}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/exception/ControllerApplicationException.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/exception/ControllerApplicationException.java
index 06d2b0e6b3f..ff61cfcc695 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/exception/ControllerApplicationException.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/exception/ControllerApplicationException.java
@@ -32,6 +32,8 @@ public class ControllerApplicationException extends
WebApplicationException {
TYPE_ONLY
}
+ private final boolean _includeCausesInResponse;
+
public ControllerApplicationException(Logger logger, String message,
Response.Status status) {
this(logger, message, status.getStatusCode(), null);
}
@@ -55,7 +57,13 @@ public class ControllerApplicationException extends
WebApplicationException {
private ControllerApplicationException(Logger logger, String message, int
status, @Nullable Throwable e,
ExceptionLogMode exceptionLogMode) {
- super(message, status);
+ this(logger, message, status, e, exceptionLogMode, false);
+ }
+
+ private ControllerApplicationException(Logger logger, String message, int
status, @Nullable Throwable e,
+ ExceptionLogMode exceptionLogMode, boolean includeCausesInResponse) {
+ super(message, includeCausesInResponse ? e : null, status);
+ _includeCausesInResponse = includeCausesInResponse;
if (status >= 300 && status < 500) {
if (e == null) {
logger.info(message);
@@ -74,4 +82,17 @@ public class ControllerApplicationException extends
WebApplicationException {
}
}
}
+
+ /// Creates an exception whose response also carries the messages of `e` and
its causes. Use it only where those
+ /// messages describe the caller's own input: the response reaches the
caller, and other causes can name files, hosts
+ /// or plugins on the controller.
+ public static ControllerApplicationException withCausesInResponse(Logger
logger, String message,
+ Response.Status status, Throwable e) {
+ return new ControllerApplicationException(logger, message,
status.getStatusCode(), e, ExceptionLogMode.FULL, true);
+ }
+
+ /// Whether the response carries the cause chain, which only
[#withCausesInResponse] asks for.
+ public boolean shouldIncludeCausesInResponse() {
+ return _includeCausesInResponse;
+ }
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotIngestionRestletResource.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotIngestionRestletResource.java
index b32880d6b94..0df2ddac56e 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotIngestionRestletResource.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotIngestionRestletResource.java
@@ -19,6 +19,7 @@
package org.apache.pinot.controller.api.resources;
import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiKeyAuthDefinition;
@@ -56,6 +57,7 @@ import
org.apache.pinot.controller.util.FileIngestionHelper.DataPayload;
import org.apache.pinot.core.auth.Actions;
import org.apache.pinot.core.auth.Authorize;
import org.apache.pinot.core.auth.TargetType;
+import
org.apache.pinot.segment.local.segment.creator.RecordProcessingException;
import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
@@ -150,13 +152,13 @@ public class PinotIngestionRestletResource {
try {
asyncResponse.resume(ingestData(tableNameWithType, batchConfigMapStr,
new DataPayload(fileUpload)));
} catch (IllegalArgumentException e) {
- asyncResponse.resume(new ControllerApplicationException(LOGGER, String
- .format("Got illegal argument when ingesting file into table: %s.
%s", tableNameWithType, e.getMessage()),
- Response.Status.BAD_REQUEST, e));
+ asyncResponse.resume(newIngestFromFileException(
+ String.format("Got illegal argument when ingesting file into table:
%s. %s", tableNameWithType,
+ e.getMessage()), Response.Status.BAD_REQUEST, e));
} catch (Exception e) {
- asyncResponse.resume(new ControllerApplicationException(LOGGER,
- String.format("Caught exception when ingesting file into table: %s.
%s", tableNameWithType, e.getMessage()),
- Response.Status.INTERNAL_SERVER_ERROR, e));
+ asyncResponse.resume(newIngestFromFileException(
+ String.format("Caught exception when ingesting file into table: %s.
%s", tableNameWithType,
+ e.getMessage()), Response.Status.INTERNAL_SERVER_ERROR, e));
}
}
@@ -236,6 +238,19 @@ public class PinotIngestionRestletResource {
return fileIngestionHelper.buildSegmentAndPush(payload);
}
+ /// A record failure returns the causes from its [RecordProcessingException]
down, which describe the uploaded data.
+ /// Wrappers above it and other failures don't: they can involve files on
the controller that the batch config names.
+ @VisibleForTesting
+ static ControllerApplicationException newIngestFromFileException(String
message, Response.Status status,
+ Exception e) {
+ RecordProcessingException recordFailure =
+ org.apache.commons.lang3.exception.ExceptionUtils.throwableOfType(e,
RecordProcessingException.class);
+ if (recordFailure != null) {
+ return ControllerApplicationException.withCausesInResponse(LOGGER,
message, status, recordFailure);
+ }
+ return new ControllerApplicationException(LOGGER, message, status, e);
+ }
+
private URI getControllerUri() {
try {
return new URI(_controllerConf.generateVipUrl());
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/WebApplicationExceptionMapper.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/WebApplicationExceptionMapper.java
index 1eb1957cd25..f0b68f2a99f 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/WebApplicationExceptionMapper.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/WebApplicationExceptionMapper.java
@@ -18,12 +18,16 @@
*/
package org.apache.pinot.controller.api.resources;
+import com.google.common.annotations.VisibleForTesting;
+import javax.annotation.Nullable;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.ext.ExceptionMapper;
import javax.ws.rs.ext.Provider;
+import org.apache.pinot.common.utils.ExceptionUtils;
import org.apache.pinot.common.utils.SimpleHttpErrorInfo;
+import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,6 +35,8 @@ import org.slf4j.LoggerFactory;
@Provider
public class WebApplicationExceptionMapper implements
ExceptionMapper<Throwable> {
private static final Logger LOGGER =
LoggerFactory.getLogger(WebApplicationExceptionMapper.class);
+ @VisibleForTesting
+ static final int MAX_CAUSES_LENGTH = 2048;
@Override
public Response toResponse(Throwable t) {
@@ -40,7 +46,18 @@ public class WebApplicationExceptionMapper implements
ExceptionMapper<Throwable>
} else {
status = ((WebApplicationException) t).getResponse().getStatus();
}
- SimpleHttpErrorInfo errorInfo = new SimpleHttpErrorInfo(status,
t.getMessage());
+ SimpleHttpErrorInfo errorInfo = new SimpleHttpErrorInfo(status,
getErrorMessage(t));
return
Response.status(status).entity(errorInfo).type(MediaType.APPLICATION_JSON).build();
}
+
+ /// Appends the cause chain only for exceptions created with
[ControllerApplicationException#withCausesInResponse].
+ @VisibleForTesting
+ @Nullable
+ static String getErrorMessage(Throwable t) {
+ if (t instanceof ControllerApplicationException
+ && ((ControllerApplicationException)
t).shouldIncludeCausesInResponse()) {
+ return ExceptionUtils.appendCauses(t.getMessage(), t.getCause(),
MAX_CAUSES_LENGTH);
+ }
+ return t.getMessage();
+ }
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotIngestionRestletResourceStatelessTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotIngestionRestletResourceStatelessTest.java
index 61251f7e6f4..cf0ef9ad920 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotIngestionRestletResourceStatelessTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotIngestionRestletResourceStatelessTest.java
@@ -42,7 +42,9 @@ import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.FileFormat;
import org.apache.pinot.spi.ingestion.batch.BatchConfigProperties;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -58,7 +60,11 @@ import static org.testng.Assert.assertTrue;
public class PinotIngestionRestletResourceStatelessTest extends ControllerTest
{
private static final String TABLE_NAME = "testTable";
private static final String TABLE_NAME_WITH_TYPE = "testTable_OFFLINE";
+ private static final String MULTI_VALUE_IN_SINGLE_VALUE_COLUMN =
"cooper;max";
+ private static final String ROOT_CAUSE =
+ "Cannot read single-value from Object[]: [cooper, max] for column: name";
private File _inputFile;
+ private File _invalidInputFile;
@BeforeClass
public void setUp()
@@ -84,6 +90,13 @@ public class PinotIngestionRestletResourceStatelessTest
extends ControllerTest {
bw.write("cat|kylo\n");
bw.write("dog|cookie\n");
}
+
+ _invalidInputFile = new File(FileUtils.getTempDirectory(),
"pinotIngestionRestletResourceTest_invalidData.csv");
+ try (BufferedWriter bw = new BufferedWriter(new
FileWriter(_invalidInputFile))) {
+ bw.write("breed|name\n");
+ bw.write("cat|kylo\n");
+ bw.write("dog|" + MULTI_VALUE_IN_SINGLE_VALUE_COLUMN + "\n");
+ }
}
@Test
@@ -141,11 +154,81 @@ public class PinotIngestionRestletResourceStatelessTest
extends ControllerTest {
assertTrue(ingestionDir.exists());
}
+ @Test(dependsOnMethods = "testIngestEndpoint", alwaysRun = true)
+ public void testIngestFromFileReturnsRootCauseOfSegmentCreationFailure()
+ throws Exception {
+ PinotAdminClient adminClient = getOrCreateAdminClient();
+ int numSegments =
_helixResourceManager.getSegmentsFor(TABLE_NAME_WITH_TYPE, false).size();
+
+ String response = sendHttpPost(
+
adminClient.getFileIngestClient().buildIngestFromFileUrl(TABLE_NAME_WITH_TYPE,
getCsvBatchConfigMap()),
+ _invalidInputFile, 500);
+
+ assertEquals(getError(response), "Caught exception when ingesting file
into table: " + TABLE_NAME_WITH_TYPE
+ + ". Caught exception while reading data -> Caught exception while
transforming data type for column: name -> "
+ + ROOT_CAUSE);
+ assertEquals(_helixResourceManager.getSegmentsFor(TABLE_NAME_WITH_TYPE,
false).size(), numSegments);
+ }
+
+ @Test(dependsOnMethods = "testIngestEndpoint", alwaysRun = true)
+ public void testIngestFromFileKeepsIllegalArgumentMessage()
+ throws Exception {
+ PinotAdminClient adminClient = getOrCreateAdminClient();
+ Map<String, String> batchConfigMap = getCsvBatchConfigMap();
+ batchConfigMap.put(BatchConfigProperties.INPUT_FORMAT, "unknownFormat");
+
+ String response = sendHttpPost(
+
adminClient.getFileIngestClient().buildIngestFromFileUrl(TABLE_NAME_WITH_TYPE,
batchConfigMap), _inputFile,
+ 400);
+
+ assertEquals(getError(response), "Got illegal argument when ingesting file
into table: " + TABLE_NAME_WITH_TYPE
+ + ". No enum constant " + FileFormat.class.getCanonicalName() +
".UNKNOWNFORMAT");
+ }
+
+ @Test(dependsOnMethods = "testIngestEndpoint", alwaysRun = true)
+ public void testIngestFromUriDoesNotReturnCauseChain()
+ throws Exception {
+ PinotAdminClient adminClient = getOrCreateAdminClient();
+ int numSegments =
_helixResourceManager.getSegmentsFor(TABLE_NAME_WITH_TYPE, false).size();
+
+
_controllerConfig.setProperty(ControllerConf.INGEST_FROM_URI_ALLOW_LOCAL_FILE_SYSTEM,
true);
+ String response;
+ try {
+ response = sendHttpPost(adminClient.getFileIngestClient()
+ .buildIngestFromUriUrl(TABLE_NAME_WITH_TYPE, getCsvBatchConfigMap(),
+ String.format("file://%s",
_invalidInputFile.getAbsolutePath())), 500);
+ } finally {
+
_controllerConfig.setProperty(ControllerConf.INGEST_FROM_URI_ALLOW_LOCAL_FILE_SYSTEM,
false);
+ }
+
+ assertEquals(getError(response), "Failed to ingest from URI");
+ assertFalse(response.contains(_invalidInputFile.getName()), response);
+ assertFalse(response.contains("cooper"), response);
+ assertEquals(_helixResourceManager.getSegmentsFor(TABLE_NAME_WITH_TYPE,
false).size(), numSegments);
+ }
+
+ private static Map<String, String> getCsvBatchConfigMap() {
+ Map<String, String> batchConfigMap = new HashMap<>();
+ batchConfigMap.put(BatchConfigProperties.INPUT_FORMAT, "csv");
+ batchConfigMap.put(String.format("%s.delimiter",
BatchConfigProperties.RECORD_READER_PROP_PREFIX), "|");
+ return batchConfigMap;
+ }
+
+ private static String getError(String responseBody)
+ throws IOException {
+ return JsonUtils.stringToJsonNode(responseBody).get("error").asText();
+ }
+
private String sendHttpPost(String uri, int expectedStatusCode)
throws IOException {
+ return sendHttpPost(uri, _inputFile, expectedStatusCode);
+ }
+
+ private String sendHttpPost(String uri, File file, int expectedStatusCode)
+ throws IOException {
HttpPost httpPost = new HttpPost(uri);
HttpEntity reqEntity =
- MultipartEntityBuilder.create().addPart("file", new
FileBody(_inputFile.getAbsoluteFile())).build();
+ MultipartEntityBuilder.create().addPart("file", new
FileBody(file.getAbsoluteFile())).build();
httpPost.setEntity(reqEntity);
try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
return httpClient.execute(httpPost, response -> {
@@ -165,6 +248,7 @@ public class PinotIngestionRestletResourceStatelessTest
extends ControllerTest {
@AfterClass
public void tearDown() {
FileUtils.deleteQuietly(_inputFile);
+ FileUtils.deleteQuietly(_invalidInputFile);
stopFakeInstances();
stopController();
stopZk();
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/exception/ControllerApplicationExceptionTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/exception/ControllerApplicationExceptionTest.java
index 539344854d2..8a0ac7d95f8 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/exception/ControllerApplicationExceptionTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/exception/ControllerApplicationExceptionTest.java
@@ -27,9 +27,13 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
-/// Tests exception logging modes used by Controller API responses.
+/// Tests exception logging modes used by Controller API responses, and which
exceptions keep their cause for them.
public class ControllerApplicationExceptionTest {
private static final String PUBLIC_MESSAGE = "Request failed";
private static final String SENSITIVE_MESSAGE =
"file:///private/controller-secret.csv?signature=secret";
@@ -61,4 +65,34 @@ public class ControllerApplicationExceptionTest {
verify(logger).error("{} exception type: {}", PUBLIC_MESSAGE,
IllegalStateException.class.getName());
verifyNoMoreInteractions(logger);
}
+
+ @Test
+ public void testConstructorsKeepTheCauseOutOfTheResponse() {
+ Logger logger = mock(Logger.class);
+ IllegalStateException cause = new IllegalStateException(SENSITIVE_MESSAGE);
+
+ ControllerApplicationException exception = new
ControllerApplicationException(logger, PUBLIC_MESSAGE,
+ Response.Status.INTERNAL_SERVER_ERROR, cause);
+
+ assertNull(exception.getCause());
+ assertFalse(exception.shouldIncludeCausesInResponse());
+ verify(logger).error(PUBLIC_MESSAGE, cause);
+ verifyNoMoreInteractions(logger);
+ }
+
+ @Test
+ public void testWithCausesInResponseKeepsTheCauseAndLogsItAsUsual() {
+ Logger logger = mock(Logger.class);
+ IllegalArgumentException cause = new IllegalArgumentException("Cannot read
single-value for column: name");
+
+ ControllerApplicationException exception =
+ ControllerApplicationException.withCausesInResponse(logger,
PUBLIC_MESSAGE, Response.Status.BAD_REQUEST, cause);
+
+ assertSame(exception.getCause(), cause);
+ assertTrue(exception.shouldIncludeCausesInResponse());
+ assertEquals(exception.getMessage(), PUBLIC_MESSAGE);
+ assertEquals(exception.getResponse().getStatus(),
Response.Status.BAD_REQUEST.getStatusCode());
+ verify(logger).info("{} exception: {}", PUBLIC_MESSAGE,
cause.getMessage());
+ verifyNoMoreInteractions(logger);
+ }
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotIngestionRestletResourceTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotIngestionRestletResourceTest.java
new file mode 100644
index 00000000000..6fa91d1e510
--- /dev/null
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/PinotIngestionRestletResourceTest.java
@@ -0,0 +1,140 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.api.resources;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.StringUtils;
+import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
+import
org.apache.pinot.segment.local.segment.creator.RecordProcessingException;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests which /ingestFromFile failures expose their causes.
+public class PinotIngestionRestletResourceTest {
+ private static final String MESSAGE = "Caught exception when ingesting file
into table: testTable_OFFLINE. ";
+ private static final String ROOT_CAUSE = "Cannot read single-value from
Object[]: [cooper, max] for column: name";
+
+ @Test
+ public void testRecordFailureIsDescribedByItsCauseChain() {
+ Exception e = new RecordProcessingException("Caught exception while
reading data",
+ new RuntimeException("Caught exception while transforming data type
for column: name",
+ new IllegalArgumentException(ROOT_CAUSE)));
+
+ ControllerApplicationException failure = getFailure(e);
+
+ assertSame(failure.getCause(), e);
+ assertEquals(WebApplicationExceptionMapper.getErrorMessage(failure),
MESSAGE + "Caught exception while reading data"
+ + " -> Caught exception while transforming data type for column: name
-> " + ROOT_CAUSE);
+ }
+
+ @Test
+ public void testWrappedRecordFailureIsDescribedFromTheRecordFailureDown() {
+ Exception recordFailure = new RecordProcessingException("Error occurred
while reading row during indexing",
+ new IllegalArgumentException(ROOT_CAUSE));
+ Exception e = new IllegalStateException("Failed to build segment",
+ new RuntimeException("Failed in working dir
/var/pinot/controller/ingestion/abc", recordFailure));
+
+ ControllerApplicationException failure = getFailure(e);
+
+ assertSame(failure.getCause(), recordFailure);
+ assertEquals(WebApplicationExceptionMapper.getErrorMessage(failure),
+ MESSAGE + "Failed to build segment -> Error occurred while reading row
during indexing -> " + ROOT_CAUSE);
+ }
+
+ @Test
+ public void testFailureToSetUpARecordReaderIsDescribedByItsMessageAlone() {
+ Exception e = new IOException("Failed to create Protobuf descriptor",
+ new FileNotFoundException("/etc/controller-secret.desc (Permission
denied)"));
+
+ ControllerApplicationException failure = getFailure(e);
+
+ assertNull(failure.getCause());
+ String message = WebApplicationExceptionMapper.getErrorMessage(failure);
+ assertEquals(message, MESSAGE + "Failed to create Protobuf descriptor");
+ assertFalse(message.contains("controller-secret"), message);
+ }
+
+ @Test
+ public void
testFailureWithoutRecordFailureInItsChainIsDescribedByItsMessageAlone() {
+ Exception e = new RuntimeException("Failed to upload segment", new
IOException("Connection reset"));
+
+ assertNull(getFailure(e).getCause());
+ assertEquals(WebApplicationExceptionMapper.getErrorMessage(getFailure(e)),
MESSAGE + "Failed to upload segment");
+ }
+
+ @Test
+ public void
testIllegalArgumentWithoutRecordFailureIsDescribedByItsMessageAlone() {
+ Exception e = new IllegalArgumentException("Invalid record reader config",
+ new FileNotFoundException("/etc/controller-secret.desc (No such file
or directory)"));
+
+ ControllerApplicationException failure =
+ PinotIngestionRestletResource.newIngestFromFileException(MESSAGE +
e.getMessage(), Response.Status.BAD_REQUEST,
+ e);
+
+ assertNull(failure.getCause());
+ assertEquals(failure.getResponse().getStatus(), 400);
+ assertEquals(WebApplicationExceptionMapper.getErrorMessage(failure),
MESSAGE + "Invalid record reader config");
+ }
+
+ @Test(timeOut = 10_000)
+ public void
testCyclicChainWithoutRecordFailureIsDescribedByItsMessageAlone() {
+ Exception first = new RuntimeException("first");
+ Exception second = new IllegalStateException("second");
+ first.initCause(second);
+ second.initCause(first);
+
+ assertNull(getFailure(first).getCause());
+ }
+
+ @Test
+ public void testFailureKeepsItsStatus() {
+ Exception e = new RecordProcessingException("Caught exception while
reading data", new IOException("inner"));
+
+
assertEquals(PinotIngestionRestletResource.newIngestFromFileException(MESSAGE,
Response.Status.BAD_REQUEST, e)
+ .getResponse().getStatus(), 400);
+ assertEquals(getFailure(new
IOException("setup")).getResponse().getStatus(), 500);
+ }
+
+ @Test
+ public void testRecordFailureIsBoundedInLength() {
+ String longValue = StringUtils.repeat('x', 10_000);
+ Exception e = new RecordProcessingException("Caught exception while
reading data",
+ new IllegalArgumentException("Cannot read single-value from Object[]:
[" + longValue + "] for column: name"));
+
+ String message =
WebApplicationExceptionMapper.getErrorMessage(getFailure(e));
+
+ String prefix = MESSAGE + "Caught exception while reading data";
+ assertTrue(message.startsWith(prefix + " -> Cannot read single-value from
Object[]: ["), message);
+ assertTrue(message.endsWith("] for column: name"), message);
+ assertEquals(message.length(), prefix.length() +
WebApplicationExceptionMapper.MAX_CAUSES_LENGTH);
+ }
+
+ private static ControllerApplicationException getFailure(Exception e) {
+ return PinotIngestionRestletResource.newIngestFromFileException(MESSAGE +
e.getMessage(),
+ Response.Status.INTERNAL_SERVER_ERROR, e);
+ }
+}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/WebApplicationExceptionMapperTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/WebApplicationExceptionMapperTest.java
new file mode 100644
index 00000000000..a4c2c5e8797
--- /dev/null
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/resources/WebApplicationExceptionMapperTest.java
@@ -0,0 +1,98 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.api.resources;
+
+import java.io.IOException;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.common.utils.SimpleHttpErrorInfo;
+import
org.apache.pinot.controller.api.exception.ControllerApplicationException;
+import org.slf4j.Logger;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests that [WebApplicationExceptionMapper] appends the cause chain only
for exceptions created to return it.
+public class WebApplicationExceptionMapperTest {
+ private static final Logger LOGGER = mock(Logger.class);
+ private static final String MESSAGE = "Caught exception when ingesting file
into table: testTable_OFFLINE";
+ private static final String ROOT_CAUSE = "Cannot read single-value from
Object[]: [cooper, max] for column: name";
+
+ @Test
+ public void
testExceptionCreatedWithCausesInResponseIsDescribedByItsCauseChain() {
+ Exception cause =
+ new RuntimeException("Caught exception while reading data", new
IllegalArgumentException(ROOT_CAUSE));
+ ControllerApplicationException e =
ControllerApplicationException.withCausesInResponse(LOGGER, MESSAGE,
+ Response.Status.INTERNAL_SERVER_ERROR, cause);
+
+ Response response = new WebApplicationExceptionMapper().toResponse(e);
+
+ assertEquals(response.getStatus(), 500);
+ SimpleHttpErrorInfo errorInfo = (SimpleHttpErrorInfo) response.getEntity();
+ assertEquals(errorInfo.getCode(), 500);
+ assertEquals(errorInfo.getError(), MESSAGE + " -> Caught exception while
reading data -> " + ROOT_CAUSE);
+ }
+
+ @Test
+ public void testCauseThatTheMessageAlreadyContainsIsNotRepeated() {
+ Exception cause =
+ new RuntimeException("Caught exception while reading data", new
IllegalArgumentException(ROOT_CAUSE));
+ ControllerApplicationException e =
ControllerApplicationException.withCausesInResponse(LOGGER,
+ MESSAGE + ". " + cause.getMessage(),
Response.Status.INTERNAL_SERVER_ERROR, cause);
+
+ assertEquals(WebApplicationExceptionMapper.getErrorMessage(e),
+ MESSAGE + ". Caught exception while reading data -> " + ROOT_CAUSE);
+ }
+
+ @Test
+ public void testCausesAppendedToTheMessageAreBounded() {
+ String rootCause =
+ "Cannot read single-value from Object[]: [" + StringUtils.repeat('x',
10_000) + "] for column: v";
+ ControllerApplicationException e =
ControllerApplicationException.withCausesInResponse(LOGGER, MESSAGE,
+ Response.Status.BAD_REQUEST, new RuntimeException("wrapper", new
IllegalArgumentException(rootCause)));
+
+ String error = WebApplicationExceptionMapper.getErrorMessage(e);
+
+ assertEquals(error.length(), MESSAGE.length() +
WebApplicationExceptionMapper.MAX_CAUSES_LENGTH);
+ assertTrue(error.startsWith(MESSAGE + " -> wrapper -> Cannot read
single-value"), error);
+ assertTrue(error.endsWith("] for column: v"), error);
+ }
+
+ /// Every other failure is described by its message alone, as before.
+ @Test
+ public void testOtherFailuresAreDescribedByTheirMessageOnly() {
+ Throwable[] failures = {
+ new ControllerApplicationException(LOGGER, MESSAGE,
Response.Status.INTERNAL_SERVER_ERROR,
+ new IllegalArgumentException(ROOT_CAUSE)),
+ new WebApplicationException(new NumberFormatException("For input
string: \"abc\""), 404),
+ new IllegalStateException("Failed to rebalance table", new
IOException("Connection reset"))
+ };
+ for (Throwable failure : failures) {
+ SimpleHttpErrorInfo errorInfo = (SimpleHttpErrorInfo) new
WebApplicationExceptionMapper().toResponse(failure)
+ .getEntity();
+ assertEquals(errorInfo.getError(), failure.getMessage(),
failure.toString());
+ }
+ assertNull(WebApplicationExceptionMapper.getErrorMessage(new
NullPointerException()));
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/RecordProcessingException.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/RecordProcessingException.java
new file mode 100644
index 00000000000..f188d7f4361
--- /dev/null
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/RecordProcessingException.java
@@ -0,0 +1,29 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.segment.creator;
+
+/// Thrown when segment creation fails to read or process a record of its
input, as opposed to failing to set up.
+public class RecordProcessingException extends RuntimeException {
+
+ private static final long serialVersionUID = 1L;
+
+ public RecordProcessingException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/RecordReaderSegmentCreationDataSource.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/RecordReaderSegmentCreationDataSource.java
index cebf83617ee..43efc29c541 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/RecordReaderSegmentCreationDataSource.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/RecordReaderSegmentCreationDataSource.java
@@ -70,7 +70,7 @@ public class RecordReaderSegmentCreationDataSource implements
SegmentCreationDat
}
} catch (Exception e) {
if (!continueOnError) {
- throw new RuntimeException("Caught exception while reading data",
e);
+ throw new RecordProcessingException("Caught exception while
reading data", e);
}
LOGGER.debug("Caught exception while reading data", e);
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentIndexCreationDriverImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentIndexCreationDriverImpl.java
index ae0120f122f..2a933b8de0a 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentIndexCreationDriverImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentIndexCreationDriverImpl.java
@@ -34,6 +34,7 @@ import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.utils.ExceptionUtils;
import
org.apache.pinot.segment.local.segment.creator.ColumnarSegmentCreationDataSource;
+import
org.apache.pinot.segment.local.segment.creator.RecordProcessingException;
import
org.apache.pinot.segment.local.segment.creator.RecordReaderSegmentCreationDataSource;
import org.apache.pinot.segment.local.segment.creator.TransformPipeline;
import
org.apache.pinot.segment.local.segment.readers.CompactedPinotSegmentRecordReader;
@@ -286,7 +287,7 @@ public class SegmentIndexCreationDriverImpl implements
SegmentIndexCreationDrive
_totalRecordReadTimeNs += recordReadStopTimeNs -
recordReadStartTimeNs;
} catch (Exception e) {
if (!_continueOnError) {
- throw new RuntimeException("Error occurred while reading row
during indexing", e);
+ throw new RecordProcessingException("Error occurred while reading
row during indexing", e);
} else {
_incompleteRowsFound++;
LOGGER.debug("Error occurred while reading row during indexing",
e);
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/RecordProcessingExceptionTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/RecordProcessingExceptionTest.java
new file mode 100644
index 00000000000..57fb6167653
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/RecordProcessingExceptionTest.java
@@ -0,0 +1,143 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.segment.creator;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordReader;
+import org.apache.pinot.spi.data.readers.RecordReaderConfig;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.expectThrows;
+
+
+/// Tests that segment creation wraps record failures in
[RecordProcessingException].
+public class RecordProcessingExceptionTest {
+ private static final File OUT_DIR = new File(FileUtils.getTempDirectory(),
"RecordProcessingExceptionTest");
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String COLUMN = "name";
+
+ @AfterMethod
+ public void tearDown() {
+ FileUtils.deleteQuietly(OUT_DIR);
+ }
+
+ @Test
+ public void testRecordThatCannotBeTransformedWhileGatheringStats()
+ throws Exception {
+ GenericRow validRow = new GenericRow();
+ validRow.putValue(COLUMN, "kylo");
+ GenericRow invalidRow = new GenericRow();
+ invalidRow.putValue(COLUMN, new Object[]{"cooper", "max"});
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(getSegmentGeneratorConfig(), new
GenericRowRecordReader(List.of(validRow, invalidRow)));
+
+ RecordProcessingException e =
expectThrows(RecordProcessingException.class, driver::build);
+
+ assertEquals(e.getMessage(), "Caught exception while reading data");
+ assertEquals(ExceptionUtils.getRootCause(e).getMessage(),
+ "Cannot read single-value from Object[]: [cooper, max] for column: " +
COLUMN);
+ }
+
+ @Test
+ public void testRecordThatCannotBeReadWhileIndexing()
+ throws Exception {
+ GenericRow row = new GenericRow();
+ row.putValue(COLUMN, "kylo");
+ IOException readFailure = new IOException("Failed to read record");
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(getSegmentGeneratorConfig(), new
FailsAfterFirstPassRecordReader(List.of(row), readFailure));
+
+ RecordProcessingException e =
expectThrows(RecordProcessingException.class, driver::build);
+
+ assertEquals(e.getMessage(), "Error occurred while reading row during
indexing");
+ assertSame(e.getCause(), readFailure);
+ }
+
+ private static SegmentGeneratorConfig getSegmentGeneratorConfig() {
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME)
+ .addSingleValueDimension(COLUMN, DataType.STRING)
+ .build();
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig,
schema);
+ config.setOutDir(OUT_DIR.getAbsolutePath());
+ config.setSegmentName("testSegment");
+ return config;
+ }
+
+ private static class FailsAfterFirstPassRecordReader implements RecordReader
{
+ private final GenericRowRecordReader _delegate;
+ private final IOException _failure;
+ private boolean _firstPassDone;
+
+ FailsAfterFirstPassRecordReader(List<GenericRow> rows, IOException
failure) {
+ _delegate = new GenericRowRecordReader(rows);
+ _failure = failure;
+ }
+
+ @Override
+ public void init(File dataFile, @Nullable Set<String> fieldsToRead,
+ @Nullable RecordReaderConfig recordReaderConfig) {
+ }
+
+ @Override
+ public boolean hasNext() {
+ boolean hasNext = _delegate.hasNext();
+ if (!hasNext) {
+ _firstPassDone = true;
+ }
+ return hasNext;
+ }
+
+ @Override
+ public GenericRow next(GenericRow reuse)
+ throws IOException {
+ if (_firstPassDone) {
+ throw _failure;
+ }
+ return _delegate.next(reuse);
+ }
+
+ @Override
+ public void rewind() {
+ _delegate.rewind();
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
