You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2018/01/16 18:27:56 UTC
[2/5] nifi git commit: NIFI-4428: Replaced JSON input with
RecordReader using per-record flowfiles NIFI-4428: Added initial L&N,
fixed dependency hierarchy
http://git-wip-us.apache.org/repos/asf/nifi/blob/ecb80678/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/java/org/apache/nifi/processors/PutDruid.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/java/org/apache/nifi/processors/PutDruid.java b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/java/org/apache/nifi/processors/PutDruid.java
deleted file mode 100644
index 3134d5a..0000000
--- a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/java/org/apache/nifi/processors/PutDruid.java
+++ /dev/null
@@ -1,196 +0,0 @@
-/*
- * 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.nifi.processors;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.nifi.annotation.behavior.InputRequirement;
-import org.apache.nifi.annotation.documentation.CapabilityDescription;
-import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.ProcessSession;
-import org.apache.nifi.processor.ProcessSessionFactory;
-import org.apache.nifi.processor.ProcessorInitializationContext;
-import org.apache.nifi.processor.Relationship;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.stream.io.StreamUtils;
-
-import org.codehaus.jackson.map.ObjectMapper;
-
-import org.apache.nifi.controller.api.DruidTranquilityService;
-import com.metamx.tranquility.tranquilizer.MessageDroppedException;
-import com.metamx.tranquility.tranquilizer.Tranquilizer;
-import com.twitter.util.Await;
-import com.twitter.util.Future;
-import com.twitter.util.FutureEventListener;
-
-import scala.runtime.BoxedUnit;
-
-@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
-@Tags({"Druid", "Timeseries", "OLAP", "ingest"})
-@CapabilityDescription("Sends events to Apache Druid for Indexing. "
- + "Leverages Druid Tranquility Controller service."
- + "Incoming flow files are expected to contain 1 or many JSON objects, one JSON object per line")
-public class PutDruid extends AbstractSessionFactoryProcessor {
-
- private List<PropertyDescriptor> properties;
- private Set<Relationship> relationships;
- private final Map<Object, String> messageStatus = new HashMap<>();
-
- public static final PropertyDescriptor DRUID_TRANQUILITY_SERVICE = new PropertyDescriptor.Builder()
- .name("putdruid-tranquility-service")
- .displayName("Tranquility Service")
- .description("Tranquility Service to use for sending events to Druid")
- .required(true)
- .identifiesControllerService(DruidTranquilityService.class)
- .build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder()
- .name("success")
- .description("Success relationship")
- .build();
-
- public static final Relationship REL_FAIL = new Relationship.Builder()
- .name("failure")
- .description("FlowFiles are routed to this relationship when they cannot be parsed")
- .build();
-
- public static final Relationship REL_DROPPED = new Relationship.Builder()
- .name("dropped")
- .description("FlowFiles are routed to this relationship when they are outside of the configured time window, timestamp format is invalid, ect...")
- .build();
-
- public void init(final ProcessorInitializationContext context) {
- List<PropertyDescriptor> properties = new ArrayList<>();
- properties.add(DRUID_TRANQUILITY_SERVICE);
- this.properties = Collections.unmodifiableList(properties);
-
- Set<Relationship> relationships = new HashSet<>();
- relationships.add(REL_SUCCESS);
- relationships.add(REL_DROPPED);
- relationships.add(REL_FAIL);
- this.relationships = Collections.unmodifiableSet(relationships);
- }
-
- @Override
- public Set<Relationship> getRelationships() {
- return relationships;
- }
-
- @Override
- protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return properties;
- }
-
- //Method breaks down incoming flow file and sends it to Druid Indexing service
- private void processFlowFile(ProcessContext context, ProcessSession session) {
- final ComponentLog log = getLogger();
-
- //Get handle on Druid Tranquility session
- DruidTranquilityService tranquilityController = context.getProperty(DRUID_TRANQUILITY_SERVICE)
- .asControllerService(DruidTranquilityService.class);
- Tranquilizer<Map<String, Object>> tranquilizer = tranquilityController.getTranquilizer();
-
- final FlowFile flowFile = session.get();
- if (flowFile == null || flowFile.getSize() == 0) {
- return;
- }
-
- //Get data from flow file body
- final byte[] buffer = new byte[(int) flowFile.getSize()];
- session.read(flowFile, in -> StreamUtils.fillBuffer(in, buffer));
-
-
- String contentString = new String(buffer, StandardCharsets.UTF_8);
- Map<String, Object> contentMap = null;
-
- //Create payload array from flow file content, one element per line
- String[] messageArray = contentString.split("\\R");
-
- //Convert each array element from JSON to HashMap and send to Druid
- for (String message : messageArray) {
- try {
- contentMap = new ObjectMapper().readValue(message, HashMap.class);
- } catch (IOException e1) {
- log.error("Error parsing incoming message array in the flowfile body");
- }
-
- log.debug("Tranquilizer Status: " + tranquilizer.status().toString());
- messageStatus.put(flowFile, "pending");
- //Send data element to Druid, Asynch
- Future<BoxedUnit> future = tranquilizer.send(contentMap);
- log.debug(" Sent Payload to Druid: " + contentMap);
-
- //Wait for Druid to call back with status
- future.addEventListener(new FutureEventListener<Object>() {
- @Override
- public void onFailure(Throwable cause) {
- if (cause instanceof MessageDroppedException) {
- //This happens when event timestamp targets a Druid Indexing task that has closed (Late Arriving Data)
- log.error(" FlowFile Dropped due to MessageDroppedException: " + cause.getMessage() + " : " + cause);
- cause.getStackTrace();
- log.error(" Transferring FlowFile to DROPPED relationship");
- session.transfer(flowFile, REL_DROPPED);
- } else {
- log.error("FlowFile Processing Failed due to: {} : {}", new Object[]{cause.getMessage(), cause});
- cause.printStackTrace();
- log.error(" Transferring FlowFile to FAIL relationship");
- session.transfer(flowFile, REL_FAIL);
- }
- }
-
- @Override
- public void onSuccess(Object value) {
- log.debug(" FlowFile Processing Success : " + value.toString());
- session.transfer(flowFile, REL_SUCCESS);
- session.getProvenanceReporter().send(flowFile, "Druid Tranquility Service");
- }
- });
-
- try {
- //Wait for result from Druid
- //This method will be asynch since this is a SessionFactoryProcessor and OnTrigger will create a new Thread
- Await.result(future);
- } catch (Exception e) {
- getLogger().error("Caught exception while waiting for result of put request: " + e.getMessage());
- }
- }
- //session.transfer(flowFile, REL_SUCCESS);
- session.commit();
- }
-
- public void onTrigger(ProcessContext context, ProcessSessionFactory factory) throws ProcessException {
- final ProcessSession session = factory.createSession();
- //Create new Thread to ensure that waiting for callback does not reduce throughput
- new Thread(() -> {
- processFlowFile(context, session);
- }).start();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/ecb80678/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/java/org/apache/nifi/processors/druid/PutDruidRecord.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/java/org/apache/nifi/processors/druid/PutDruidRecord.java b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/java/org/apache/nifi/processors/druid/PutDruidRecord.java
new file mode 100644
index 0000000..19e47a9
--- /dev/null
+++ b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/java/org/apache/nifi/processors/druid/PutDruidRecord.java
@@ -0,0 +1,373 @@
+/*
+ * 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.nifi.processors.druid;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.metamx.tranquility.tranquilizer.MessageDroppedException;
+import com.twitter.util.Future;
+import com.twitter.util.FutureEventListener;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+
+import org.apache.nifi.controller.api.druid.DruidTranquilityService;
+import com.metamx.tranquility.tranquilizer.Tranquilizer;
+import org.apache.nifi.serialization.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+import scala.runtime.BoxedUnit;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"druid", "timeseries", "olap", "ingest", "put", "record"})
+@CapabilityDescription("Sends records to Druid for Indexing. Leverages Druid Tranquility Controller service.")
+@WritesAttribute(attribute = "record.count", description = "The number of messages that were sent to Druid for this FlowFile. FlowFiles on the success relationship will have a value "
+ + "of this attribute that indicates the number of records successfully processed by Druid, and the FlowFile content will be only the successful records. This behavior applies "
+ + "to the failure and dropped relationships as well.")
+public class PutDruidRecord extends AbstractSessionFactoryProcessor {
+
+ static final String RECORD_COUNT = "record.count";
+
+ private List<PropertyDescriptor> properties;
+ private Set<Relationship> relationships;
+
+ static final PropertyDescriptor RECORD_READER_FACTORY = new PropertyDescriptor.Builder()
+ .name("putdruid-record-reader")
+ .displayName("Record Reader")
+ .description("Specifies the Controller Service to use for parsing incoming data and determining the data's schema.")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor RECORD_WRITER_FACTORY = new PropertyDescriptor.Builder()
+ .name("putdruid-record-writer")
+ .displayName("Record Writer")
+ .description("The Record Writer to use in order to serialize the data to outgoing relationships.")
+ .identifiesControllerService(RecordSetWriterFactory.class)
+ .expressionLanguageSupported(false)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor DRUID_TRANQUILITY_SERVICE = new PropertyDescriptor.Builder()
+ .name("putdruid-tranquility-service")
+ .displayName("Tranquility Service")
+ .description("Tranquility Service to use for sending events to Druid.")
+ .required(true)
+ .identifiesControllerService(DruidTranquilityService.class)
+ .build();
+
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("FlowFiles are routed to this relationship when they are successfully processed by Druid")
+ .build();
+
+ static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("FlowFiles are routed to this relationship when they cannot be parsed or otherwise processed by Druid")
+ .build();
+
+ static final Relationship REL_DROPPED = new Relationship.Builder()
+ .name("dropped")
+ .description("FlowFiles are routed to this relationship when they are outside of the configured time window, timestamp format is invalid, ect...")
+ .build();
+
+
+ public void init(final ProcessorInitializationContext context) {
+ List<PropertyDescriptor> properties = new ArrayList<>();
+ properties.add(RECORD_READER_FACTORY);
+ properties.add(RECORD_WRITER_FACTORY);
+ properties.add(DRUID_TRANQUILITY_SERVICE);
+ this.properties = Collections.unmodifiableList(properties);
+
+ Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_DROPPED);
+ relationships.add(REL_FAILURE);
+ this.relationships = Collections.unmodifiableSet(relationships);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return properties;
+ }
+
+ /**
+ * Parses the record(s), converts each to a Map, and sends via Tranquility to the Druid Indexing Service
+ *
+ * @param context The process context
+ * @param session The process session
+ */
+ @SuppressWarnings("unchecked")
+ private void processFlowFile(ProcessContext context, final ProcessSession session) {
+ final ComponentLog log = getLogger();
+
+ // Get handle on Druid Tranquility session
+ DruidTranquilityService tranquilityController = context.getProperty(DRUID_TRANQUILITY_SERVICE)
+ .asControllerService(DruidTranquilityService.class);
+ Tranquilizer<Map<String, Object>> tranquilizer = tranquilityController.getTranquilizer();
+
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ // Create the outgoing flow files and output streams
+ FlowFile droppedFlowFile = session.create(flowFile);
+ final AtomicInteger droppedFlowFileCount = new AtomicInteger(0);
+ FlowFile failedFlowFile = session.create(flowFile);
+ final AtomicInteger failedFlowFileCount = new AtomicInteger(0);
+ FlowFile successfulFlowFile = session.create(flowFile);
+ final AtomicInteger successfulFlowFileCount = new AtomicInteger(0);
+
+ final AtomicInteger recordWriteErrors = new AtomicInteger(0);
+
+ int recordCount = 0;
+ final OutputStream droppedOutputStream = session.write(droppedFlowFile);
+ final RecordSetWriter droppedRecordWriter;
+ final OutputStream failedOutputStream = session.write(failedFlowFile);
+ final RecordSetWriter failedRecordWriter;
+ final OutputStream successfulOutputStream = session.write(successfulFlowFile);
+ final RecordSetWriter successfulRecordWriter;
+ try (final InputStream in = session.read(flowFile)) {
+
+ final RecordReaderFactory recordParserFactory = context.getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER_FACTORY).asControllerService(RecordSetWriterFactory.class);
+
+ final Map<String, String> attributes = flowFile.getAttributes();
+
+ final RecordReader reader = recordParserFactory.createRecordReader(flowFile, in, getLogger());
+ final RecordSchema outSchema = writerFactory.getSchema(attributes, reader.getSchema());
+ droppedRecordWriter = writerFactory.createWriter(log, outSchema, droppedOutputStream);
+ droppedRecordWriter.beginRecordSet();
+ failedRecordWriter = writerFactory.createWriter(log, outSchema, failedOutputStream);
+ failedRecordWriter.beginRecordSet();
+ successfulRecordWriter = writerFactory.createWriter(log, outSchema, successfulOutputStream);
+ successfulRecordWriter.beginRecordSet();
+
+ Record r;
+ while ((r = reader.nextRecord()) != null) {
+ final Record record = r;
+ recordCount++;
+ // Convert each Record to HashMap and send to Druid
+ Map<String, Object> contentMap = (Map<String, Object>) DataTypeUtils.convertRecordFieldtoObject(r, RecordFieldType.RECORD.getRecordDataType(r.getSchema()));
+
+ log.debug("Tranquilizer Status: {}", new Object[]{tranquilizer.status().toString()});
+ // Send data element to Druid asynchronously
+ Future<BoxedUnit> future = tranquilizer.send(contentMap);
+ log.debug("Sent Payload to Druid: {}", new Object[]{contentMap});
+
+ // Wait for Druid to call back with status
+ future.addEventListener(new FutureEventListener<Object>() {
+ @Override
+ public void onFailure(Throwable cause) {
+ if (cause instanceof MessageDroppedException) {
+ // This happens when event timestamp targets a Druid Indexing task that has closed (Late Arriving Data)
+ log.debug("Record Dropped due to MessageDroppedException: {}, transferring record to dropped.", new Object[]{cause.getMessage()}, cause);
+ try {
+ synchronized (droppedRecordWriter) {
+ droppedRecordWriter.write(record);
+ droppedRecordWriter.flush();
+ droppedFlowFileCount.incrementAndGet();
+ }
+ } catch (final IOException ioe) {
+ log.error("Error transferring record to dropped, this may result in data loss.", new Object[]{ioe.getMessage()}, ioe);
+ recordWriteErrors.incrementAndGet();
+ }
+
+ } else {
+ log.error("FlowFile Processing Failed due to: {}", new Object[]{cause.getMessage()}, cause);
+ try {
+ synchronized (failedRecordWriter) {
+ failedRecordWriter.write(record);
+ failedRecordWriter.flush();
+ failedFlowFileCount.incrementAndGet();
+ }
+ } catch (final IOException ioe) {
+ log.error("Error transferring record to failure, this may result in data loss.", new Object[]{ioe.getMessage()}, ioe);
+ recordWriteErrors.incrementAndGet();
+ }
+ }
+ }
+
+ @Override
+ public void onSuccess(Object value) {
+ log.debug(" FlowFile Processing Success: {}", new Object[]{value.toString()});
+ try {
+ synchronized (successfulRecordWriter) {
+ successfulRecordWriter.write(record);
+ successfulRecordWriter.flush();
+ successfulFlowFileCount.incrementAndGet();
+ }
+ } catch (final IOException ioe) {
+ log.error("Error transferring record to success, this may result in data loss. "
+ + "However the record was successfully processed by Druid", new Object[]{ioe.getMessage()}, ioe);
+ recordWriteErrors.incrementAndGet();
+ }
+ }
+ });
+
+ }
+
+ } catch (IOException | SchemaNotFoundException | MalformedRecordException e) {
+ log.error("FlowFile Processing Failed due to: {}", new Object[]{e.getMessage()}, e);
+ // The FlowFile will be obtained and the error logged below, when calling publishResult.getFailedFlowFiles()
+ flowFile = session.putAttribute(flowFile, RECORD_COUNT, Integer.toString(recordCount));
+ session.transfer(flowFile, REL_FAILURE);
+ try {
+ droppedOutputStream.close();
+ session.remove(droppedFlowFile);
+ } catch (IOException ioe) {
+ log.error("Error closing output stream for FlowFile with dropped records.", ioe);
+ }
+ try {
+ failedOutputStream.close();
+ session.remove(failedFlowFile);
+ } catch (IOException ioe) {
+ log.error("Error closing output stream for FlowFile with failed records.", ioe);
+ }
+ try {
+ successfulOutputStream.close();
+ session.remove(successfulFlowFile);
+ } catch (IOException ioe) {
+ log.error("Error closing output stream for FlowFile with successful records.", ioe);
+ }
+ session.commit();
+ return;
+ }
+
+ if (recordCount == 0) {
+ // Send original (empty) flow file to success, remove the rest
+ flowFile = session.putAttribute(flowFile, RECORD_COUNT, "0");
+ session.transfer(flowFile, REL_SUCCESS);
+ try {
+ droppedOutputStream.close();
+ session.remove(droppedFlowFile);
+ } catch (IOException ioe) {
+ log.error("Error closing output stream for FlowFile with dropped records.", ioe);
+ }
+ try {
+ failedOutputStream.close();
+ session.remove(failedFlowFile);
+ } catch (IOException ioe) {
+ log.error("Error closing output stream for FlowFile with failed records.", ioe);
+ }
+ try {
+ successfulOutputStream.close();
+ session.remove(successfulFlowFile);
+ } catch (IOException ioe) {
+ log.error("Error closing output stream for FlowFile with successful records.", ioe);
+ }
+ } else {
+
+ // Wait for all the records to finish processing
+ while (recordCount != (droppedFlowFileCount.get() + failedFlowFileCount.get() + successfulFlowFileCount.get() + recordWriteErrors.get())) {
+ Thread.yield();
+ }
+
+ // Send partitioned flow files out to their relationships (or remove them if empty)
+
+ try {
+ droppedRecordWriter.finishRecordSet();
+ droppedRecordWriter.close();
+ } catch (IOException ioe) {
+ log.error("Error closing FlowFile with dropped records: {}", new Object[]{ioe.getMessage()}, ioe);
+ session.rollback();
+ throw new ProcessException(ioe);
+ }
+ if (droppedFlowFileCount.get() > 0) {
+ droppedFlowFile = session.putAttribute(droppedFlowFile, RECORD_COUNT, Integer.toString(droppedFlowFileCount.get()));
+ session.transfer(droppedFlowFile, REL_DROPPED);
+ } else {
+ session.remove(droppedFlowFile);
+ }
+
+ try {
+ failedRecordWriter.finishRecordSet();
+ failedRecordWriter.close();
+ } catch (IOException ioe) {
+ log.error("Error closing FlowFile with failed records: {}", new Object[]{ioe.getMessage()}, ioe);
+ session.rollback();
+ throw new ProcessException(ioe);
+ }
+ if (failedFlowFileCount.get() > 0) {
+ failedFlowFile = session.putAttribute(failedFlowFile, RECORD_COUNT, Integer.toString(failedFlowFileCount.get()));
+ session.transfer(failedFlowFile, REL_FAILURE);
+ } else {
+ session.remove(failedFlowFile);
+ }
+
+ try {
+ successfulRecordWriter.finishRecordSet();
+ successfulRecordWriter.close();
+ } catch (IOException ioe) {
+ log.error("Error closing FlowFile with successful records: {}", new Object[]{ioe.getMessage()}, ioe);
+ session.rollback();
+ throw new ProcessException(ioe);
+ }
+ if (successfulFlowFileCount.get() > 0) {
+ successfulFlowFile = session.putAttribute(successfulFlowFile, RECORD_COUNT, Integer.toString(successfulFlowFileCount.get()));
+ session.transfer(successfulFlowFile, REL_SUCCESS);
+ session.getProvenanceReporter().send(successfulFlowFile, tranquilityController.getTransitUri());
+ } else {
+ session.remove(successfulFlowFile);
+ }
+
+ session.remove(flowFile);
+ }
+
+ session.commit();
+ }
+
+ public void onTrigger(ProcessContext context, ProcessSessionFactory factory) throws ProcessException {
+ final ProcessSession session = factory.createSession();
+ processFlowFile(context, session);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/ecb80678/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index 4136d5e..25ef747 100644
--- a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++ b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -13,4 +13,4 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-org.apache.nifi.processors.PutDruid
\ No newline at end of file
+org.apache.nifi.processors.druid.PutDruidRecord
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/ecb80678/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/controller/druid/MockDruidTranquilityController.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/controller/druid/MockDruidTranquilityController.java b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/controller/druid/MockDruidTranquilityController.java
new file mode 100644
index 0000000..695212c
--- /dev/null
+++ b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/controller/druid/MockDruidTranquilityController.java
@@ -0,0 +1,150 @@
+/*
+ * 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.nifi.controller.druid;
+
+import com.metamx.tranquility.beam.Beam;
+import com.metamx.tranquility.tranquilizer.MessageDroppedException;
+import com.metamx.tranquility.tranquilizer.Tranquilizer;
+import com.metamx.tranquility.typeclass.Timestamper;
+import com.twitter.finagle.Status;
+import com.twitter.util.Awaitable;
+import com.twitter.util.Duration;
+import com.twitter.util.Future;
+import com.twitter.util.FutureEventListener;
+import com.twitter.util.TimeoutException;
+import com.twitter.util.Try;
+import io.druid.data.input.impl.TimestampSpec;
+import io.druid.query.aggregation.AggregatorFactory;
+import org.apache.curator.framework.CuratorFramework;
+import scala.Function1;
+import scala.Option;
+import scala.runtime.BoxedUnit;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class MockDruidTranquilityController extends DruidTranquilityController {
+
+ private final Tranquilizer t;
+ private final CuratorFramework cf;
+ private int numCalls = 0;
+
+ public MockDruidTranquilityController() {
+ this(-1, -1);
+ }
+
+ /**
+ * Creates a mock/stub Druid controller for testing. The failAfterN parameter must be higher than the dropAfterN parameter in order for messages to be dropped.
+ *
+ * @param dropAfterN The number of records after which to start calling the "dropped" callback, namely onFailure(MessageDroppedException)
+ * @param failAfterN The number of records after which to start calling the "failure" callback, namely onFailure(Exception)
+ */
+ public MockDruidTranquilityController(final int dropAfterN, final int failAfterN) {
+ t = mock(Tranquilizer.class);
+ final Future<BoxedUnit> future = new Future<BoxedUnit>() {
+
+ FutureEventListener<? super BoxedUnit> listener;
+
+ @Override
+ public Future<BoxedUnit> addEventListener(FutureEventListener<? super BoxedUnit> listener) {
+ this.listener = listener;
+ numCalls++;
+ if (dropAfterN >= 0 && numCalls > failAfterN) {
+ listener.onFailure(new Exception());
+ } else if (dropAfterN >= 0 && numCalls > dropAfterN) {
+ listener.onFailure(MessageDroppedException.Instance());
+ } else {
+ listener.onSuccess(BoxedUnit.UNIT);
+ }
+ return this;
+ }
+
+ @Override
+ public Awaitable<BoxedUnit> ready(Duration timeout, CanAwait permit) throws InterruptedException, TimeoutException {
+ return null;
+ }
+
+ @Override
+ public BoxedUnit result(Duration timeout, CanAwait permit) throws Exception {
+ return null;
+ }
+
+ @Override
+ public boolean isReady(CanAwait permit) {
+ return true;
+ }
+
+ @Override
+ public Future<BoxedUnit> respond(Function1<Try<BoxedUnit>, BoxedUnit> k) {
+ return null;
+ }
+
+ @Override
+ public Option<Try<BoxedUnit>> poll() {
+ return null;
+ }
+
+ @Override
+ public void raise(Throwable interrupt) {
+
+ }
+
+ @Override
+ public <B> Future<B> transform(Function1<Try<BoxedUnit>, Future<B>> f) {
+ return null;
+ }
+ };
+ when(t.send(anyObject())).thenReturn(future);
+ when(t.status()).thenReturn(new Status() {
+ });
+ cf = mock(CuratorFramework.class);
+ }
+
+ @Override
+ public Tranquilizer getTranquilizer() {
+ return t;
+ }
+
+ @Override
+ CuratorFramework getCurator(String zkConnectString) {
+ return cf;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ Tranquilizer<Map<String, Object>> buildTranquilizer(int maxBatchSize, int maxPendingBatches, int lingerMillis, Beam<Map<String, Object>> beam) {
+ return t;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ Beam<Map<String, Object>> buildBeam(String dataSource, String indexService, String discoveryPath, int clusterPartitions, int clusterReplication,
+ String segmentGranularity, String queryGranularity, String windowPeriod, String indexRetryPeriod, List<String> dimensions,
+ List<AggregatorFactory> aggregator, Timestamper<Map<String, Object>> timestamper, TimestampSpec timestampSpec) {
+ return mock(Beam.class);
+ }
+
+ @Override
+ public String getTransitUri() {
+ return "";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ecb80678/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/processors/PutDruidTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/processors/PutDruidTest.java b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/processors/PutDruidTest.java
deleted file mode 100644
index 0137624..0000000
--- a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/processors/PutDruidTest.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.nifi.processors;
-
-
-public class PutDruidTest {
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/ecb80678/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/processors/druid/PutDruidRecordTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/processors/druid/PutDruidRecordTest.java b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/processors/druid/PutDruidRecordTest.java
new file mode 100644
index 0000000..28e084c
--- /dev/null
+++ b/nifi-nar-bundles/nifi-druid-bundle/nifi-druid-processors/src/test/java/org/apache/nifi/processors/druid/PutDruidRecordTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.nifi.processors.druid;
+
+
+import org.apache.nifi.controller.druid.DruidTranquilityController;
+import org.apache.nifi.controller.druid.MockDruidTranquilityController;
+import org.apache.nifi.controller.api.druid.DruidTranquilityService;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.serialization.record.MockRecordParser;
+import org.apache.nifi.serialization.record.MockRecordWriter;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+
+public class PutDruidRecordTest {
+
+ private TestRunner runner;
+ private DruidTranquilityService druidTranquilityController;
+ private MockRecordParser recordReader;
+ private MockRecordWriter recordWriter;
+
+ @Before
+ public void setUp() throws Exception {
+ runner = TestRunners.newTestRunner(PutDruidRecord.class);
+ druidTranquilityController = new MockDruidTranquilityController(2,3);
+ recordReader = new MockRecordParser();
+ recordWriter = new MockRecordWriter(null, true, 2);
+ runner.addControllerService("reader", recordReader);
+ runner.enableControllerService(recordReader);
+ runner.addControllerService("writer", recordWriter);
+ runner.enableControllerService(recordWriter);
+
+ runner.addControllerService("tranquility", druidTranquilityController);
+ runner.setProperty(druidTranquilityController, DruidTranquilityController.DATASOURCE, "test");
+ runner.setProperty(druidTranquilityController, DruidTranquilityController.ZOOKEEPER_CONNECTION_STRING, "localhost:2181");
+ runner.setProperty(druidTranquilityController, DruidTranquilityController.AGGREGATOR_JSON, "[{\"type\": \"count\", \"name\": \"count\"}]");
+ runner.setProperty(druidTranquilityController, DruidTranquilityController.DIMENSIONS_LIST, "dim1,dim2");
+ runner.assertValid(druidTranquilityController);
+ runner.enableControllerService(druidTranquilityController);
+
+ runner.setProperty(PutDruidRecord.RECORD_READER_FACTORY, "reader");
+ runner.setProperty(PutDruidRecord.RECORD_WRITER_FACTORY, "writer");
+ runner.setProperty(PutDruidRecord.DRUID_TRANQUILITY_SERVICE, "tranquility");
+ }
+
+ @Test
+ public void testEmptyRecord() throws Exception {
+ recordReader.addSchemaField("name", RecordFieldType.STRING);
+ recordReader.addSchemaField("age", RecordFieldType.INT);
+ recordReader.addSchemaField("sport", RecordFieldType.STRING);
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertTransferCount(PutDruidRecord.REL_SUCCESS, 1);
+ runner.assertTransferCount(PutDruidRecord.REL_FAILURE, 0);
+ runner.assertTransferCount(PutDruidRecord.REL_DROPPED, 0);
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(PutDruidRecord.REL_SUCCESS).get(0);
+ flowFile.assertAttributeEquals(PutDruidRecord.RECORD_COUNT, "0");
+ }
+
+ @Test
+ public void testPutRecords() throws Exception {
+ recordReader.addSchemaField("name", RecordFieldType.STRING);
+ recordReader.addSchemaField("age", RecordFieldType.INT);
+ recordReader.addSchemaField("sport", RecordFieldType.STRING);
+
+ recordReader.addRecord("John Doe", 48, "Soccer");
+ recordReader.addRecord("Jane Doe", 47, "Tennis");
+ recordReader.addRecord("Sally Doe", 47, "Curling"); // Will be dropped due to the "drop after 2" parameter on the MockDruidTranquilityController
+ recordReader.addRecord("Jimmy Doe", 14, null); // Will fail due to the "fail after 3" parameter on the MockDruidTranquilityController
+ recordReader.addRecord("Pizza Doe", 14, null); // Will fail due to the "fail after 3" parameter on the MockDruidTranquilityController
+ recordReader.addRecord("Bad Record", "X", 13); // RecordWriter fail due to the "fail after 2" parameter on the MockRecordWriter, not written to output
+
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertTransferCount(PutDruidRecord.REL_SUCCESS, 1);
+ runner.assertTransferCount(PutDruidRecord.REL_FAILURE, 1);
+ runner.assertTransferCount(PutDruidRecord.REL_DROPPED, 1);
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(PutDruidRecord.REL_SUCCESS).get(0);
+ flowFile.assertAttributeEquals(PutDruidRecord.RECORD_COUNT, "2");
+ flowFile = runner.getFlowFilesForRelationship(PutDruidRecord.REL_DROPPED).get(0);
+ flowFile.assertAttributeEquals(PutDruidRecord.RECORD_COUNT, "1");
+ flowFile = runner.getFlowFilesForRelationship(PutDruidRecord.REL_FAILURE).get(0);
+ flowFile.assertAttributeEquals(PutDruidRecord.RECORD_COUNT, "2");
+
+ // Assert a single SEND event present for the successful flow file
+ assertEquals(1, runner.getProvenanceEvents().stream().filter((e) -> ProvenanceEventType.SEND.equals(e.getEventType())).count());
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi/blob/ecb80678/nifi-nar-bundles/nifi-druid-bundle/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-druid-bundle/pom.xml b/nifi-nar-bundles/nifi-druid-bundle/pom.xml
index 6b19faf..82d3017 100644
--- a/nifi-nar-bundles/nifi-druid-bundle/pom.xml
+++ b/nifi-nar-bundles/nifi-druid-bundle/pom.xml
@@ -27,10 +27,25 @@
<packaging>pom</packaging>
<properties>
- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- <maven.compiler.source>1.8</maven.compiler.source>
- <maven.compiler.target>1.8</maven.compiler.target>
+ <druid.version>0.9.1</druid.version>
+ <tranquility.version>0.8.2</tranquility.version>
</properties>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>io.druid</groupId>
+ <artifactId>tranquility-core_2.11</artifactId>
+ <version>${tranquility.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.druid</groupId>
+ <artifactId>druid-processing</artifactId>
+ <version>${druid.version}</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
<modules>
<module>nifi-druid-nar</module>
<module>nifi-druid-controller-service-api-nar</module>
http://git-wip-us.apache.org/repos/asf/nifi/blob/ecb80678/nifi-nar-bundles/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/pom.xml b/nifi-nar-bundles/pom.xml
index 46eb2e3..2632a4a 100755
--- a/nifi-nar-bundles/pom.xml
+++ b/nifi-nar-bundles/pom.xml
@@ -91,6 +91,7 @@
<module>nifi-metrics-reporting-bundle</module>
<module>nifi-spark-bundle</module>
<module>nifi-atlas-bundle</module>
+ <module>nifi-druid-bundle</module>
</modules>
<build>