Repository: nifi
Updated Branches:
  refs/heads/master 1c58e78ce -> b1901d5fe


http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordSetWriterFactory.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordSetWriterFactory.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordSetWriterFactory.java
index e23ad20..9ea6e64 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordSetWriterFactory.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-service-api/src/main/java/org/apache/nifi/serialization/RecordSetWriterFactory.java
@@ -24,6 +24,7 @@ import org.apache.nifi.controller.ControllerService;
 import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.record.RecordSchema;
 
 /**
  * <p>
@@ -45,19 +46,29 @@ public interface RecordSetWriterFactory extends 
ControllerService {
 
     /**
      * <p>
-     * Creates a new RecordSetWriter that is capable of writing record 
contents to an OutputStream. Note that the
-     * FlowFile and InputStream that are given may well be different than the 
FlowFile that the writer is intended
-     * to write to. The given FlowFile and InputStream are intended to be used 
for determining the schema that should
-     * be used when writing records.
+     * Returns the Schema that will be used for writing Records. Note that the 
FlowFile and InputStream that are given
+     * may well be different than the FlowFile that the writer will write to. 
The given FlowFile and InputStream are
+     * intended to be used for determining the schema that should be used when 
writing records.
+     * </p>
+     *
+     * @param flowFile the FlowFile from which the schema should be determined.
+     * @param content the contents of the FlowFile from which to determine the 
schema
+     * @return the Schema that should be used for writing Records
+     * @throws SchemaNotFoundException if unable to find the schema
+     */
+    RecordSchema getSchema(FlowFile flowFile, InputStream content) throws 
SchemaNotFoundException, IOException;
+
+    /**
+     * <p>
+     * Creates a new RecordSetWriter that is capable of writing record 
contents to an OutputStream.
      * </p>
      *
      * @param logger the logger to use when logging information. This is 
passed in, rather than using the logger of the Controller Service
      *            because it allows messages to be logged for the component 
that is calling this Controller Service.
-     * @param schemaFlowFile the FlowFile from which the schema should be 
determined.
-     * @param schemaFlowFileContent the contents of the FlowFile from which to 
determine the schema
+     * @param schema the schema that will be used for writing records
+     *
      * @return a RecordSetWriter that can write record sets to an OutputStream
-     * @throws SchemaNotFoundException if unable to find the schema
      * @throws IOException if unable to read from the given InputStream
      */
-    RecordSetWriter createWriter(ComponentLog logger, FlowFile schemaFlowFile, 
InputStream schemaFlowFileContent) throws SchemaNotFoundException, IOException;
+    RecordSetWriter createWriter(ComponentLog logger, RecordSchema schema) 
throws SchemaNotFoundException, IOException;
 }

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroReaderWithExplicitSchema.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroReaderWithExplicitSchema.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroReaderWithExplicitSchema.java
index daabdcb..ce49443 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroReaderWithExplicitSchema.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroReaderWithExplicitSchema.java
@@ -33,7 +33,6 @@ import org.apache.nifi.serialization.record.RecordSchema;
 
 public class AvroReaderWithExplicitSchema extends AvroRecordReader {
     private final InputStream in;
-    private final Schema avroSchema;
     private final RecordSchema recordSchema;
     private final DatumReader<GenericRecord> datumReader;
     private final BinaryDecoder decoder;
@@ -43,7 +42,6 @@ public class AvroReaderWithExplicitSchema extends 
AvroRecordReader {
         this.in = in;
         this.recordSchema = recordSchema;
 
-        this.avroSchema = avroSchema;
         datumReader = new GenericDatumReader<GenericRecord>(avroSchema);
         decoder = DecoderFactory.get().binaryDecoder(in, null);
     }

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroRecordSetWriter.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroRecordSetWriter.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroRecordSetWriter.java
index a4476f9..1360add 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroRecordSetWriter.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/avro/AvroRecordSetWriter.java
@@ -18,7 +18,6 @@
 package org.apache.nifi.avro;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.LinkedHashMap;
@@ -32,7 +31,6 @@ import 
org.apache.nifi.annotation.documentation.CapabilityDescription;
 import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.components.AllowableValue;
 import org.apache.nifi.components.ValidationContext;
-import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.processor.exception.ProcessException;
 import org.apache.nifi.schema.access.SchemaField;
@@ -59,12 +57,10 @@ public class AvroRecordSetWriter extends 
SchemaRegistryRecordSetWriter implement
         "The FlowFile will have the Avro schema embedded into the content, as 
is typical with Avro");
 
     @Override
-    public RecordSetWriter createWriter(final ComponentLog logger, final 
FlowFile flowFile, final InputStream in) throws IOException {
+    public RecordSetWriter createWriter(final ComponentLog logger, final 
RecordSchema recordSchema) throws IOException {
         final String strategyValue = 
getConfigurationContext().getProperty(getSchemaWriteStrategyDescriptor()).getValue();
 
         try {
-            final RecordSchema recordSchema = getSchema(flowFile, in);
-
             final Schema avroSchema;
             try {
                 if (recordSchema.getSchemaFormat().isPresent() & 
recordSchema.getSchemaFormat().get().equals(AvroTypeUtil.AVRO_SCHEMA_FORMAT)) {

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
index 45b5bae..3bcbce2 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordReader.java
@@ -112,7 +112,7 @@ public class CSVRecordReader implements RecordReader {
             return null;
         }
 
-        return DataTypeUtils.convertType(trimmed, dataType, dateFormat, 
timeFormat, timestampFormat, fieldName);
+        return DataTypeUtils.convertType(trimmed, dataType, () -> dateFormat, 
() -> timeFormat, () -> timestampFormat, fieldName);
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
index 0b29f09..d4f066f 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVRecordSetWriter.java
@@ -18,7 +18,6 @@
 package org.apache.nifi.csv;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -28,7 +27,6 @@ import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.annotation.lifecycle.OnEnabled;
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.controller.ConfigurationContext;
-import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.schema.access.SchemaNotFoundException;
 import org.apache.nifi.serialization.DateTimeTextRecordSetWriter;
@@ -69,8 +67,7 @@ public class CSVRecordSetWriter extends 
DateTimeTextRecordSetWriter implements R
     }
 
     @Override
-    public RecordSetWriter createWriter(final ComponentLog logger, final 
FlowFile flowFile, final InputStream in) throws SchemaNotFoundException, 
IOException {
-        final RecordSchema schema = getSchema(flowFile, in);
+    public RecordSetWriter createWriter(final ComponentLog logger, final 
RecordSchema schema) throws SchemaNotFoundException, IOException {
         return new WriteCSVResult(csvFormat, schema, 
getSchemaAccessWriter(schema),
             getDateFormat().orElse(null), getTimeFormat().orElse(null), 
getTimestampFormat().orElse(null), includeHeader);
     }

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
index 7d2a2c1..1ee35ba 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonPathRowRecordReader.java
@@ -162,7 +162,7 @@ public class JsonPathRowRecordReader extends 
AbstractJsonRowRecordReader {
 
             return new MapRecord(childSchema, coercedValues);
         } else {
-            return DataTypeUtils.convertType(value, dataType, dateFormat, 
timeFormat, timestampFormat, fieldName);
+            return DataTypeUtils.convertType(value, dataType, () -> 
dateFormat, () -> timeFormat, () -> timestampFormat, fieldName);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
index 163b2ec..91370a7 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
@@ -18,7 +18,6 @@
 package org.apache.nifi.json;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -27,7 +26,6 @@ import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.annotation.lifecycle.OnEnabled;
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.controller.ConfigurationContext;
-import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.schema.access.SchemaNotFoundException;
 import org.apache.nifi.serialization.DateTimeTextRecordSetWriter;
@@ -64,8 +62,7 @@ public class JsonRecordSetWriter extends 
DateTimeTextRecordSetWriter implements
     }
 
     @Override
-    public RecordSetWriter createWriter(final ComponentLog logger, final 
FlowFile flowFile, final InputStream flowFileContent) throws 
SchemaNotFoundException, IOException {
-        final RecordSchema schema = getSchema(flowFile, flowFileContent);
+    public RecordSetWriter createWriter(final ComponentLog logger, final 
RecordSchema schema) throws SchemaNotFoundException, IOException {
         return new WriteJsonResult(logger, schema, 
getSchemaAccessWriter(schema), prettyPrint,
             getDateFormat().orElse(null), getTimeFormat().orElse(null), 
getTimestampFormat().orElse(null));
     }

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
index ee5bff9..b542ebe 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
@@ -127,13 +127,14 @@ public class JsonTreeRowRecordReader extends 
AbstractJsonRowRecordReader {
             case SHORT:
                 return DataTypeUtils.toShort(getRawNodeValue(fieldNode), 
fieldName);
             case STRING:
-                return DataTypeUtils.toString(getRawNodeValue(fieldNode), 
DataTypeUtils.getDateFormat(desiredType.getFieldType(), dateFormat, timeFormat, 
timestampFormat));
+                return DataTypeUtils.toString(getRawNodeValue(fieldNode),
+                    () -> 
DataTypeUtils.getDateFormat(desiredType.getFieldType(), () -> dateFormat, () -> 
timeFormat, () -> timestampFormat));
             case DATE:
-                return DataTypeUtils.toDate(getRawNodeValue(fieldNode), 
dateFormat, fieldName);
+                return DataTypeUtils.toDate(getRawNodeValue(fieldNode), () -> 
dateFormat, fieldName);
             case TIME:
-                return DataTypeUtils.toTime(getRawNodeValue(fieldNode), 
timeFormat, fieldName);
+                return DataTypeUtils.toTime(getRawNodeValue(fieldNode), () -> 
timeFormat, fieldName);
             case TIMESTAMP:
-                return DataTypeUtils.toTimestamp(getRawNodeValue(fieldNode), 
timestampFormat, fieldName);
+                return DataTypeUtils.toTimestamp(getRawNodeValue(fieldNode), 
() -> timestampFormat, fieldName);
             case MAP: {
                 final DataType valueType = ((MapDataType) 
desiredType).getValueType();
 

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
index ac4fbee..bd7dc5e 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/WriteJsonResult.java
@@ -162,7 +162,7 @@ public class WriteJsonResult implements RecordSetWriter {
         }
 
         final DataType chosenDataType = dataType.getFieldType() == 
RecordFieldType.CHOICE ? DataTypeUtils.chooseDataType(value, (ChoiceDataType) 
dataType) : dataType;
-        final Object coercedValue = DataTypeUtils.convertType(value, 
chosenDataType, dateFormat, timeFormat, timestampFormat, fieldName);
+        final Object coercedValue = DataTypeUtils.convertType(value, 
chosenDataType, () -> dateFormat, () -> timeFormat, () -> timestampFormat, 
fieldName);
         if (coercedValue == null) {
             generator.writeNull();
             return;
@@ -170,7 +170,7 @@ public class WriteJsonResult implements RecordSetWriter {
 
         switch (chosenDataType.getFieldType()) {
             case DATE: {
-                final String stringValue = 
DataTypeUtils.toString(coercedValue, dateFormat);
+                final String stringValue = 
DataTypeUtils.toString(coercedValue, () -> dateFormat);
                 if (DataTypeUtils.isLongTypeCompatible(stringValue)) {
                     generator.writeNumber(DataTypeUtils.toLong(coercedValue, 
fieldName));
                 } else {
@@ -179,7 +179,7 @@ public class WriteJsonResult implements RecordSetWriter {
                 break;
             }
             case TIME: {
-                final String stringValue = 
DataTypeUtils.toString(coercedValue, timeFormat);
+                final String stringValue = 
DataTypeUtils.toString(coercedValue, () -> timeFormat);
                 if (DataTypeUtils.isLongTypeCompatible(stringValue)) {
                     generator.writeNumber(DataTypeUtils.toLong(coercedValue, 
fieldName));
                 } else {
@@ -188,7 +188,7 @@ public class WriteJsonResult implements RecordSetWriter {
                 break;
             }
             case TIMESTAMP: {
-                final String stringValue = 
DataTypeUtils.toString(coercedValue, timestampFormat);
+                final String stringValue = 
DataTypeUtils.toString(coercedValue, () -> timestampFormat);
                 if (DataTypeUtils.isLongTypeCompatible(stringValue)) {
                     generator.writeNumber(DataTypeUtils.toLong(coercedValue, 
fieldName));
                 } else {

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/text/FreeFormTextRecordSetWriter.java
----------------------------------------------------------------------
diff --git 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/text/FreeFormTextRecordSetWriter.java
 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/text/FreeFormTextRecordSetWriter.java
index cb69444..022804e 100644
--- 
a/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/text/FreeFormTextRecordSetWriter.java
+++ 
b/nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/text/FreeFormTextRecordSetWriter.java
@@ -17,7 +17,6 @@
 
 package org.apache.nifi.text;
 
-import java.io.InputStream;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
@@ -27,19 +26,19 @@ import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.annotation.lifecycle.OnEnabled;
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.components.PropertyValue;
-import org.apache.nifi.controller.AbstractControllerService;
 import org.apache.nifi.controller.ConfigurationContext;
-import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.serialization.RecordSetWriter;
 import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.SchemaRegistryRecordSetWriter;
+import org.apache.nifi.serialization.record.RecordSchema;
 
 @Tags({"text", "freeform", "expression", "language", "el", "record", 
"recordset", "resultset", "writer", "serialize"})
 @CapabilityDescription("Writes the contents of a RecordSet as free-form text. 
The configured "
     + "text is able to make use of the Expression Language to reference each 
of the fields that are available "
     + "in a Record. Each record in the RecordSet will be separated by a single 
newline character.")
-public class FreeFormTextRecordSetWriter extends AbstractControllerService 
implements RecordSetWriterFactory {
+public class FreeFormTextRecordSetWriter extends SchemaRegistryRecordSetWriter 
implements RecordSetWriterFactory {
     static final PropertyDescriptor TEXT = new PropertyDescriptor.Builder()
         .name("Text")
         .description("The text to use when writing the results. This property 
will evaluate the Expression Language using any of the fields available in a 
Record.")
@@ -74,8 +73,7 @@ public class FreeFormTextRecordSetWriter extends 
AbstractControllerService imple
     }
 
     @Override
-    public RecordSetWriter createWriter(final ComponentLog logger, final 
FlowFile flowFile, final InputStream in) {
+    public RecordSetWriter createWriter(final ComponentLog logger, final 
RecordSchema schema) {
         return new FreeFormTextWriter(textValue, characterSet);
     }
-
 }

http://git-wip-us.apache.org/repos/asf/nifi/blob/b1901d5f/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 57b1835..45d0cc3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1459,6 +1459,11 @@
             </dependency>
             <dependency>
                 <groupId>org.apache.nifi</groupId>
+                <artifactId>nifi-record-path</artifactId>
+                <version>1.3.0-SNAPSHOT</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.nifi</groupId>
                 <artifactId>nifi-avro-record-utils</artifactId>
                 <version>1.3.0-SNAPSHOT</version>
             </dependency>

Reply via email to