[GitHub] [nifi] YolandaMDavis commented on a change in pull request #4020: NIFI-7054: Add RecordSinkServiceLookup for dynamic sink routing

[GitBox <gi...@apache.org>]

YolandaMDavis commented on a change in pull request #4020: NIFI-7054: Add RecordSinkServiceLookup for dynamic sink routing
URL: https://github.com/apache/nifi/pull/4020#discussion_r372744657
 
 

 ##########
 File path: nifi-nar-bundles/nifi-standard-services/nifi-record-sink-service-bundle/nifi-record-sink-service/src/main/java/org/apache/nifi/record/sink/lookup/RecordSinkServiceLookup.java
 ##########
 @@ -0,0 +1,83 @@
+/*
+ * 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.record.sink.lookup;
+
+import org.apache.nifi.annotation.behavior.DynamicProperty;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.record.sink.RecordSinkService;
+import org.apache.nifi.serialization.WriteResult;
+import org.apache.nifi.serialization.record.RecordSet;
+import org.apache.nifi.service.lookup.AbstractSingleAttributeBasedControllerServiceLookup;
+
+import java.io.IOException;
+import java.util.Map;
+
+
+@Tags({"record", "sink", "lookup"})
+@CapabilityDescription("Provides a RecordSinkService that can be used to dynamically select another RecordSinkService. This service " +
+        "requires an attribute named 'record.sink.name' to be passed in when asking for a connection, and will throw an exception " +
+        "if the attribute is missing. The value of 'record.sink.name' will be used to select the RecordSinkService that has been " +
+        "registered with that name. This will allow multiple RecordSinkServices to be defined and registered, and then selected " +
+        "dynamically at runtime by tagging flow files with the appropriate 'record.sink.name' attribute.")
+@DynamicProperty(name = "The name to register the specified RecordSinkService", value = "The RecordSinkService",
+        description = "If '" + RecordSinkServiceLookup.RECORD_SINK_NAME_ATTRIBUTE + "' attribute contains " +
+                "the name of the dynamic property, then the RecordSinkService (registered in the value) will be selected.",
+        expressionLanguageScope = ExpressionLanguageScope.NONE)
+public class RecordSinkServiceLookup
+        extends AbstractSingleAttributeBasedControllerServiceLookup<RecordSinkService> implements RecordSinkService {
+
+    public static final String RECORD_SINK_NAME_ATTRIBUTE = "record.sink.name";
+
+    RecordSinkService recordSinkService;
+
+    @Override
+    protected String getLookupAttribute() {
+        return RECORD_SINK_NAME_ATTRIBUTE;
+    }
+
+    @Override
+    public Class<RecordSinkService> getServiceType() {
+        return RecordSinkService.class;
+    }
+
+    @Override
+    public WriteResult sendData(RecordSet recordSet, Map<String, String> attributes, boolean sendZeroResults) throws IOException {
+        try {
+            RecordSinkService recordSink = lookupService(attributes);
+            WriteResult writeResult = recordSink.sendData(recordSet, attributes, sendZeroResults);
+            if (recordSinkService == null) {
 
 Review comment:
   I think this only sets the recordSinkService variable once, however if the value changes, and if the reset is invoked, only the first used in the lookup service is reset.  Should there be a way to ensure that all used values are reset when invoked? Or at least to ensure that the recordSinkService is changed when a different recordSink is retrieved?

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services