Github user jvwing commented on a diff in the pull request: https://github.com/apache/nifi/pull/1942#discussion_r126293188 --- Diff: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/PutRethinkDB.java --- @@ -0,0 +1,237 @@ +/* + * 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.rethinkdb; + +import org.apache.nifi.annotation.behavior.EventDriven; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.SupportsBatching; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.flowfile.FlowFile; +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.json.simple.parser.JSONParser; +import com.rethinkdb.gen.ast.Insert; +import java.io.ByteArrayOutputStream; +import java.nio.charset.Charset; +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; + +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +@EventDriven +@SupportsBatching +@Tags({"rethinkdb", "stream","insert", "update", "write", "put"}) +@CapabilityDescription("Processor to write the JSON content of a FlowFile to RethinkDB (https://www.rethinkdb.com/). The flow file should contain either JSON Object an array of JSON documents") +@WritesAttributes({ + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_ERROR_MESSAGE, description = "RethinkDB error message"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_ERROR_KEY, description = "Error count while inserting documents"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_DELETED_KEY, description = "Number of documents deleted"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_GENERATED_KEYS_KEY, description = "Keys generated on inserting documents"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_INSERTED_KEY, description = "Number of documents inserted"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_REPLACED_KEY, description = "Number of documents replaced"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_SKIPPED_KEY, description = "Number of documents skipped because they already existed"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_UNCHANGED_KEY, description = "Number of documents unchanged since they already existed"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_FIRST_ERROR_KEY, description = "First error while inserting documents"), + @WritesAttribute(attribute = PutRethinkDB.RETHINKDB_INSERT_RESULT_WARNINGS_KEY, description = "Warning message in case of large number of ids being returned on insertion") + }) +public class PutRethinkDB extends AbstractRethinkDBProcessor { + + public static AllowableValue CONFLICT_STRATEGY_UPDATE = new AllowableValue("update", "Update", "Update the document having same id with new values"); + public static AllowableValue CONFLICT_STRATEGY_REPLACE = new AllowableValue("replace", "Replace", "Replace the document with having same id new document"); + public static AllowableValue CONFLICT_STRATEGY_ERROR = new AllowableValue("error", "Error", "Return error if the document with same id exists"); + + public static AllowableValue DURABILITY_SOFT = new AllowableValue("soft", "Soft", "Don't save document on disk before ack"); + public static AllowableValue DURABILITY_HARD = new AllowableValue("hard", "Hard", "Save document on disk before ack"); + + protected static final PropertyDescriptor CONFLICT_STRATEGY = new PropertyDescriptor.Builder() + .name("rethinkdb-conflict-strategy") + .displayName("Conflict strategy") + .description("Conflict strategy to be used in case of inserting existing document.") + .required(true) + .defaultValue(CONFLICT_STRATEGY_UPDATE.getValue()) + .allowableValues(CONFLICT_STRATEGY_UPDATE, CONFLICT_STRATEGY_REPLACE, CONFLICT_STRATEGY_ERROR) + .expressionLanguageSupported(true) + .build(); + + protected static final PropertyDescriptor DURABILITY = new PropertyDescriptor.Builder() + .name("rethinkdb-durability") + .displayName("Durablity of documents") + .description("Durability of documents being inserted") + .required(true) + .defaultValue("hard") + .allowableValues(DURABILITY_HARD, DURABILITY_SOFT) + .expressionLanguageSupported(true) + .build(); + + protected String CONFLICT_OPTION_KEY = "conflict"; + protected String DURABILITY_OPTION_KEY = "durability"; + + private static final Set<Relationship> relationships; + private static final List<PropertyDescriptor> propertyDescriptors; + + public static final String RETHINKDB_ERROR_MESSAGE = "rethinkdb.error.message"; + public static final String RETHINKDB_INSERT_RESULT = "rethinkdb.insert.result"; + public static final String RETHINKDB_INSERT_RESULT_ERROR_KEY = "rethinkdb.insert.errors"; + public static final String RETHINKDB_INSERT_RESULT_DELETED_KEY = "rethinkdb.insert.deleted"; + public static final String RETHINKDB_INSERT_RESULT_GENERATED_KEYS_KEY = "rethinkdb.insert.generated_keys"; + public static final String RETHINKDB_INSERT_RESULT_INSERTED_KEY = "rethinkdb.insert.inserted"; + public static final String RETHINKDB_INSERT_RESULT_REPLACED_KEY = "rethinkdb.insert.replaced"; + public static final String RETHINKDB_INSERT_RESULT_SKIPPED_KEY = "rethinkdb.insert.skipped"; + public static final String RETHINKDB_INSERT_RESULT_UNCHANGED_KEY = "rethinkdb.insert.unchanged"; + public static final String RETHINKDB_INSERT_RESULT_FIRST_ERROR_KEY = "rethinkdb.insert.first_error"; + public static final String RETHINKDB_INSERT_RESULT_WARNINGS_KEY = "rethinkdb.insert.warnings"; + + static { + final Set<Relationship> tempRelationships = new HashSet<>(); + tempRelationships.add(REL_SUCCESS); + tempRelationships.add(REL_FAILURE); + relationships = Collections.unmodifiableSet(tempRelationships); + + final List<PropertyDescriptor> tempDescriptors = new ArrayList<>(); + tempDescriptors.add(DB_NAME); + tempDescriptors.add(DB_HOST); + tempDescriptors.add(DB_PORT); + tempDescriptors.add(USERNAME); + tempDescriptors.add(PASSWORD); + tempDescriptors.add(TABLE_NAME); + tempDescriptors.add(CHARSET); + tempDescriptors.add(CONFLICT_STRATEGY); + tempDescriptors.add(DURABILITY); + tempDescriptors.add(MAX_DOCUMENTS_SIZE); + propertyDescriptors = Collections.unmodifiableList(tempDescriptors); + } + + @Override + public Set<Relationship> getRelationships() { + return relationships; + } + + @Override + public final List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return propertyDescriptors; + } + + @OnScheduled + public void onScheduled(final ProcessContext context) { + super.onScheduled(context); + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + FlowFile flowFile = session.get(); + if (flowFile == null) { + return; + } + + if ( flowFile.getSize() == 0) { + getLogger().error("Empty message"); + flowFile = session.putAttribute(flowFile, RETHINKDB_ERROR_MESSAGE, "Empty message size " + flowFile.getSize()); + session.transfer(flowFile, REL_FAILURE); + return; + } + + if ( flowFile.getSize() > maxDocumentsSize) { + getLogger().error("Message size exceeded {} max allowed is {}", new Object[] { flowFile.getSize(), maxDocumentsSize}); + flowFile = session.putAttribute(flowFile, RETHINKDB_ERROR_MESSAGE, "Max message size exceeded " + flowFile.getSize()); + session.transfer(flowFile, REL_FAILURE); + return; + } + + Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue()); + String conflictStrategy = context.getProperty(CONFLICT_STRATEGY).evaluateAttributeExpressions(flowFile).getValue(); + String durability = context.getProperty(DURABILITY).evaluateAttributeExpressions(flowFile).getValue(); + + try { + long startTimeMillis = System.currentTimeMillis(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + session.exportTo(flowFile, baos); + String documents = new String(baos.toByteArray(), charset); + JSONParser parser = new JSONParser(); + Object jsonDocuments = parser.parse(documents); + + Insert insert = getRdbTable().insert(jsonDocuments) + .optArg(CONFLICT_OPTION_KEY, conflictStrategy) + .optArg(DURABILITY_OPTION_KEY, durability); + + HashMap<String,Object> result = runInsert(insert); + final long endTimeMillis = System.currentTimeMillis(); + getLogger().debug("Json documents {} inserted Result: {}", new Object[] {documents, result}); + flowFile = populateAttributes(session, flowFile, result); + + if ( (Long)result.get(RESULT_ERROR_KEY) != 0 ) { + getLogger().error("There were errors while inserting data documents {} result {}", + new Object [] {documents, result}); + session.transfer(flowFile, REL_FAILURE); + } else { + session.transfer(flowFile, REL_SUCCESS); + session.getProvenanceReporter().send(flowFile, + new StringBuilder("rethinkdb://").append(databaseName).append("/").append(tableName).toString(), + (endTimeMillis - startTimeMillis)); + } + } catch (Exception exception) { + getLogger().error("Failed to insert into RethinkDB due to {}", + new Object[]{exception.getLocalizedMessage()}, exception); + flowFile = session.putAttribute(flowFile, RETHINKDB_ERROR_MESSAGE, exception.getMessage() + ""); --- End diff -- I notice you use the `+ ""` method to cast objects to strings in several places. Any special reason? It does not appear too much in the NiFi codebase, as opposed to `String.valueOf()`.
--- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---