You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2021/12/04 13:52:40 UTC
[GitHub] [druid] abhishekagarwal87 commented on a change in pull request #11961: Add parse error list API for stream supervisors, use structured object for parse exceptions, simplify parse exception message
abhishekagarwal87 commented on a change in pull request #11961:
URL: https://github.com/apache/druid/pull/11961#discussion_r762426171
##########
File path: core/src/main/java/org/apache/druid/java/util/common/parsers/UnparseableColumnsParseException.java
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.druid.java.util.common.parsers;
+
+import java.util.List;
+
+public class UnparseableColumnsParseException extends ParseException
Review comment:
Please add javadocs about what his class is about.
##########
File path: processing/src/main/java/org/apache/druid/segment/incremental/ParseExceptionReport.java
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.druid.segment.incremental;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.query.ordering.StringComparators;
+
+import java.util.List;
+import java.util.Objects;
+
+public class ParseExceptionReport implements Comparable<ParseExceptionReport>
+{
+ private final String input;
+ private final String errorType;
+ private final List<String> details;
+ private final long timeOfExceptionMillis;
+
+ @JsonCreator
+ public ParseExceptionReport(
+ @JsonProperty("input") String input,
+ @JsonProperty("errorType") String errorType,
+ @JsonProperty("details") List<String> details,
+ @JsonProperty("timeOfExceptionMillis") long timeOfExceptionMillis
+ )
+ {
+ this.input = input;
+ this.errorType = errorType;
+ this.details = details;
+ this.timeOfExceptionMillis = timeOfExceptionMillis;
+ }
+
+ @JsonProperty
+ public String getInput()
+ {
+ return input;
+ }
+
+ @JsonProperty
+ public String getErrorType()
+ {
+ return errorType;
+ }
+
+ @JsonProperty
+ public List<String> getDetails()
+ {
+ return details;
+ }
+
+ @JsonProperty
+ public long getTimeOfExceptionMillis()
+ {
+ return timeOfExceptionMillis;
+ }
+
+ @Override
+ public int compareTo(ParseExceptionReport o)
Review comment:
```
* <p>It is strongly recommended, but <i>not</i> strictly required that
* <tt>(x.compareTo(y)==0) == (x.equals(y))</tt>. Generally speaking, any
* class that implements the <tt>Comparable</tt> interface and violates
* this condition should clearly indicate this fact. The recommended
* language is "Note: this class has a natural ordering that is
* inconsistent with equals."
```
From `compareTo` documentation. do we not want to include `details` when doing the comparison? If not, maybe it is better to create a separate comparator class and pass it to `TreeSet` than overriding `compareTo` here.
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
##########
@@ -262,6 +263,8 @@ public TaskStatus runTask(final TaskToolbox toolbox)
inputSource,
toolbox.getIndexingTmpDir()
);
+
+ Thread.sleep(60000);
Review comment:
what is this for?
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskClient.java
##########
@@ -224,6 +231,31 @@ public DateTime getStartTime(final String id)
}
}
+ public List<ParseExceptionReport> getParseErrors(final String id)
+ {
+ log.debug("getParseErrors task[%s]", id);
+
+ try {
+ final StringFullResponseHolder response = submitRequestWithEmptyContent(
+ id,
+ HttpMethod.GET,
+ "unparseableEvents",
+ null,
+ true
+ );
+ return response.getContent() == null || response.getContent().isEmpty()
+ ? Collections.emptyList()
+ : deserialize(response.getContent(), TYPE_REFERENCE_LIST_PARSE_EXCEPTION_REPORT);
+ }
+ catch (NoTaskLocationException e) {
+ return Collections.emptyList();
Review comment:
will this be logged anywhere as a warning? I have same question for situation when respose content is null or empty.
##########
File path: extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoder.java
##########
@@ -98,35 +98,40 @@ public DynamicMessage parse(ByteBuffer bytes)
url = new URL(descriptorFilePath);
}
catch (MalformedURLException e) {
- throw new ParseException(e, "Descriptor not found in class path or malformed URL:" + descriptorFilePath);
+ throw new ParseException(
+ descriptorFilePath,
+ e,
+ "Descriptor not found in class path or malformed URL:" + descriptorFilePath
+ );
}
try {
fin = url.openConnection().getInputStream();
}
catch (IOException e) {
- throw new ParseException(e, "Cannot read descriptor file: " + url);
+ throw new ParseException(url.toString(), e, "Cannot read descriptor file: " + url);
}
}
DynamicSchema dynamicSchema;
try {
dynamicSchema = DynamicSchema.parseFrom(fin);
}
catch (Descriptors.DescriptorValidationException e) {
- throw new ParseException(e, "Invalid descriptor file: " + descriptorFilePath);
+ throw new ParseException(descriptorFilePath, e, "Invalid descriptor file: " + descriptorFilePath);
}
catch (IOException e) {
- throw new ParseException(e, "Cannot read descriptor file: " + descriptorFilePath);
+ throw new ParseException(descriptorFilePath, e, "Cannot read descriptor file: " + descriptorFilePath);
}
Set<String> messageTypes = dynamicSchema.getMessageTypes();
if (messageTypes.size() == 0) {
- throw new ParseException("No message types found in the descriptor: " + descriptorFilePath);
+ throw new ParseException(descriptorFilePath, "No message types found in the descriptor: " + descriptorFilePath);
}
String messageType = protoMessageType == null ? (String) messageTypes.toArray()[0] : protoMessageType;
Descriptors.Descriptor desc = dynamicSchema.getMessageDescriptor(messageType);
if (desc == null) {
throw new ParseException(
+ null,
Review comment:
does it make sense to use messageType instead of null?
##########
File path: core/src/main/java/org/apache/druid/java/util/common/parsers/ParseException.java
##########
@@ -38,26 +40,43 @@
public class ParseException extends RuntimeException
{
private final boolean fromPartiallyValidRow;
+ private final long timeOfExceptionMillis;
+ private final String input;
- public ParseException(String formatText, Object... arguments)
+ public ParseException(@Nullable String input, String formatText, Object... arguments)
{
super(StringUtils.nonStrictFormat(formatText, arguments));
+ this.input = input;
this.fromPartiallyValidRow = false;
+ this.timeOfExceptionMillis = System.currentTimeMillis();
}
- public ParseException(boolean fromPartiallyValidRow, String formatText, Object... arguments)
+ public ParseException(@Nullable String input, boolean fromPartiallyValidRow, String formatText, Object... arguments)
Review comment:
nitpick - can you add a comment that input is the string representation of the input data on which the ParseException is being thrown for?
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
##########
@@ -1132,6 +1168,26 @@ public Boolean isHealthy()
}
}
+ @Override
+ public List<ParseExceptionReport> getParseErrors()
+ {
+ try {
+ if (spec.getSpec().getTuningConfig().convertToTaskTuningConfig().getMaxParseExceptions() <= 0) {
+ return ImmutableList.of();
+ }
+ lastKnownParseErrors = getCurrentParseErrors();
+ return lastKnownParseErrors;
+ }
+ catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ log.error(ie, "getErrors() interrupted.");
Review comment:
```suggestion
log.error(ie, "getCurrentParseErrors() interrupted.");
```
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
##########
@@ -1200,6 +1256,85 @@ public Boolean isHealthy()
return allStats;
}
+ /**
+ * Collect parse errors from all tasks managed by this supervisor.
+ *
+ * @return A list of parse error strings
+ *
+ * @throws InterruptedException
+ * @throws ExecutionException
+ * @throws TimeoutException
+ */
+ private List<ParseExceptionReport> getCurrentParseErrors()
Review comment:
maybe I missed but have you added unit tests for this block of code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org