You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by da...@apache.org on 2015/06/07 14:31:29 UTC
[14/51] [abbrv] incubator-nifi git commit: NIFI-551 -
ConvertJSONToAvro improve error message
NIFI-551 - ConvertJSONToAvro improve error message
- Report failure counts as an log error message
- Send record parsing errors to a separate flowfile which is transfered down the
failure relationship
Signed-off-by: joewitt <jo...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/6f32e6e9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/6f32e6e9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/6f32e6e9
Branch: refs/heads/NIFI-632
Commit: 6f32e6e9776548b746f6388d25067cd17aa1898c
Parents: 8201381
Author: ricky <ri...@cloudera.com>
Authored: Tue Apr 28 15:58:09 2015 -0400
Committer: joewitt <jo...@apache.org>
Committed: Thu Jun 4 14:52:16 2015 -0400
----------------------------------------------------------------------
.../nifi/processors/kite/ConvertJSONToAvro.java | 38 ++++++++++++++------
.../kite/TestJSONToAvroProcessor.java | 23 ++++++++++--
2 files changed, 47 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f32e6e9/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java b/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java
index 78f80b9..d4cc760 100644
--- a/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java
+++ b/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java
@@ -38,6 +38,7 @@ import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.io.StreamCallback;
import org.kitesdk.data.DatasetException;
import org.kitesdk.data.DatasetIOException;
@@ -97,22 +98,22 @@ public class ConvertJSONToAvro extends AbstractKiteProcessor {
}
@Override
- public void onTrigger(ProcessContext context, final ProcessSession session)
+ public void onTrigger(final ProcessContext context, final ProcessSession session)
throws ProcessException {
- FlowFile flowFile = session.get();
- if (flowFile == null) {
+ FlowFile successfulRecords = session.get();
+ if (successfulRecords == null) {
return;
}
String schemaProperty = context.getProperty(SCHEMA)
- .evaluateAttributeExpressions(flowFile)
+ .evaluateAttributeExpressions(successfulRecords)
.getValue();
final Schema schema;
try {
schema = getSchema(schemaProperty, DefaultConfiguration.get());
} catch (SchemaNotFoundException e) {
getLogger().error("Cannot find schema: " + schemaProperty);
- session.transfer(flowFile, FAILURE);
+ session.transfer(successfulRecords, FAILURE);
return;
}
@@ -121,21 +122,31 @@ public class ConvertJSONToAvro extends AbstractKiteProcessor {
writer.setCodec(CodecFactory.snappyCodec());
try {
- flowFile = session.write(flowFile, new StreamCallback() {
+ successfulRecords = session.write(successfulRecords, new StreamCallback() {
@Override
public void process(InputStream in, OutputStream out) throws IOException {
+ FlowFile failedRecords = session.create();
long written = 0L;
long errors = 0L;
+ long total = 0L;
try (JSONFileReader<Record> reader = new JSONFileReader<>(
in, schema, Record.class)) {
reader.initialize();
try (DataFileWriter<Record> w = writer.create(schema, out)) {
while (reader.hasNext()) {
+ total += 1;
try {
Record record = reader.next();
w.append(record);
written += 1;
- } catch (DatasetRecordException e) {
+ } catch (final DatasetRecordException e) {
+ failedRecords = session.append(failedRecords, new OutputStreamCallback() {
+ @Override
+ public void process(OutputStream out) throws IOException {
+ out.write((e.getMessage() + " [" +
+ e.getCause().getMessage() + "]\n").getBytes());
+ }
+ });
errors += 1;
}
}
@@ -143,21 +154,26 @@ public class ConvertJSONToAvro extends AbstractKiteProcessor {
session.adjustCounter("Converted records", written,
false /* update only if file transfer is successful */);
session.adjustCounter("Conversion errors", errors,
- false /* update only if file transfer is successful */);
+ false /* update only if file transfer is successful */);
+
+ if (errors > 0L) {
+ getLogger().warn("Failed to convert " + errors + '/' + total + " records from JSON to Avro");
+ }
}
+ session.transfer(failedRecords, FAILURE);
}
});
- session.transfer(flowFile, SUCCESS);
+ session.transfer(successfulRecords, SUCCESS);
//session.getProvenanceReporter().send(flowFile, target.getUri().toString());
} catch (ProcessException | DatasetIOException e) {
getLogger().error("Failed reading or writing", e);
- session.transfer(flowFile, FAILURE);
+ session.transfer(successfulRecords, FAILURE);
} catch (DatasetException e) {
getLogger().error("Failed to read FlowFile", e);
- session.transfer(flowFile, FAILURE);
+ session.transfer(successfulRecords, FAILURE);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f32e6e9/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/test/java/org/apache/nifi/processors/kite/TestJSONToAvroProcessor.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/test/java/org/apache/nifi/processors/kite/TestJSONToAvroProcessor.java b/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/test/java/org/apache/nifi/processors/kite/TestJSONToAvroProcessor.java
index d50e7f9..0b53bc7 100644
--- a/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/test/java/org/apache/nifi/processors/kite/TestJSONToAvroProcessor.java
+++ b/nifi/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/test/java/org/apache/nifi/processors/kite/TestJSONToAvroProcessor.java
@@ -18,9 +18,17 @@
*/
package org.apache.nifi.processors.kite;
+import java.io.File;
+import java.io.FileInputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Path;
+import java.util.List;
+
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Assert;
@@ -38,9 +46,13 @@ public class TestJSONToAvroProcessor {
public static final String JSON_CONTENT = ""
+ "{\"id\": 1,\"color\": \"green\"}"
- + "{\"id\": \"120V\", \"color\": \"blue\"}\n" + // invalid, ID is a string
+ + "{\"id\": \"120V\", \"color\": \"blue\"}\n" // invalid, ID is a string
+ + "{\"id\": 10, \"color\": 15.23}\n" + // invalid, color as double
"{\"id\": 2, \"color\": \"grey\", \"price\": 12.95 }";
+ public static final String FAILURE_CONTENT = "Cannot convert field id [Cannot convert to long: \"120V\"]\n" +
+ "Cannot convert field color [Cannot convert to string: 15.23]\n";
+
@Test
public void testBasicConversion() throws IOException {
TestRunner runner = TestRunners.newTestRunner(ConvertJSONToAvro.class);
@@ -54,8 +66,13 @@ public class TestJSONToAvroProcessor {
long converted = runner.getCounterValue("Converted records");
long errors = runner.getCounterValue("Conversion errors");
Assert.assertEquals("Should convert 2 rows", 2, converted);
- Assert.assertEquals("Should reject 1 row", 1, errors);
+ Assert.assertEquals("Should reject 2 rows", 2, errors);
+
+ runner.assertTransferCount("success", 1);
+ runner.assertTransferCount("failure", 1);
- runner.assertAllFlowFilesTransferred("success", 1);
+ String failureContent = Bytes.toString(runner.getContentAsByteArray(
+ runner.getFlowFilesForRelationship("failure").get(0)));
+ Assert.assertEquals("Should reject an invalid string and double", FAILURE_CONTENT, failureContent);
}
}