exceptionfactory commented on code in PR #6589:
URL: https://github.com/apache/nifi/pull/6589#discussion_r1006993638
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/ml/AwsMlProcessor.java:
##########
@@ -0,0 +1,148 @@
+package org.apache.nifi.processors.aws.ml;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.AmazonWebServiceRequest;
+import com.amazonaws.AmazonWebServiceResult;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentials;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processors.aws.AbstractAWSCredentialsProviderProcessor;
+
+public abstract class AwsMlProcessor<SERVICE extends AmazonWebServiceClient,
REQUEST extends AmazonWebServiceRequest, RESPONSE extends
AmazonWebServiceResult> extends
AbstractAWSCredentialsProviderProcessor<SERVICE> {
+ protected static final String AWS_TASK_ID_PROPERTY = "awsTaskId";
+ public static final PropertyDescriptor JSON_PAYLOAD = new
PropertyDescriptor.Builder()
+ .name("json-payload")
+ .displayName("JSON Payload")
+ .description("JSON Payload that represent an AWS ML Request. See
more details in AWS API documentation.")
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .build();
+ public static final PropertyDescriptor
MANDATORY_AWS_CREDENTIALS_PROVIDER_SERVICE =
+ new
PropertyDescriptor.Builder().fromPropertyDescriptor(AWS_CREDENTIALS_PROVIDER_SERVICE)
+ .required(true)
+ .build();
+ private static final List<PropertyDescriptor> PROPERTIES =
Collections.unmodifiableList(Arrays.asList(
+ JSON_PAYLOAD,
+ MANDATORY_AWS_CREDENTIALS_PROVIDER_SERVICE,
+ TIMEOUT,
+ SSL_CONTEXT_SERVICE,
+ ENDPOINT_OVERRIDE,
+ PROXY_CONFIGURATION_SERVICE,
+ PROXY_HOST,
+ PROXY_HOST_PORT,
+ PROXY_USERNAME,
+ PROXY_PASSWORD));
Review Comment:
The individual Proxy properties are not necessary since all of that can be
configured using the Proxy Configuration Service.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/ml/AwsMlProcessor.java:
##########
@@ -0,0 +1,148 @@
+package org.apache.nifi.processors.aws.ml;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.AmazonWebServiceRequest;
+import com.amazonaws.AmazonWebServiceResult;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentials;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processors.aws.AbstractAWSCredentialsProviderProcessor;
+
+public abstract class AwsMlProcessor<SERVICE extends AmazonWebServiceClient,
REQUEST extends AmazonWebServiceRequest, RESPONSE extends
AmazonWebServiceResult> extends
AbstractAWSCredentialsProviderProcessor<SERVICE> {
+ protected static final String AWS_TASK_ID_PROPERTY = "awsTaskId";
+ public static final PropertyDescriptor JSON_PAYLOAD = new
PropertyDescriptor.Builder()
+ .name("json-payload")
+ .displayName("JSON Payload")
+ .description("JSON Payload that represent an AWS ML Request. See
more details in AWS API documentation.")
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .build();
+ public static final PropertyDescriptor
MANDATORY_AWS_CREDENTIALS_PROVIDER_SERVICE =
+ new
PropertyDescriptor.Builder().fromPropertyDescriptor(AWS_CREDENTIALS_PROVIDER_SERVICE)
+ .required(true)
+ .build();
+ private static final List<PropertyDescriptor> PROPERTIES =
Collections.unmodifiableList(Arrays.asList(
+ JSON_PAYLOAD,
+ MANDATORY_AWS_CREDENTIALS_PROVIDER_SERVICE,
+ TIMEOUT,
+ SSL_CONTEXT_SERVICE,
+ ENDPOINT_OVERRIDE,
+ PROXY_CONFIGURATION_SERVICE,
+ PROXY_HOST,
+ PROXY_HOST_PORT,
+ PROXY_USERNAME,
+ PROXY_PASSWORD));
+ private final ObjectMapper mapper = JsonMapper.builder()
+ .configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true)
+ .build();
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+ RESPONSE response;
+ try {
+ response = sendRequest(buildRequest(session, context, flowFile),
context);
+ } catch (Exception e) {
+ session.transfer(flowFile, REL_FAILURE);
+ getLogger().error("Exception was thrown during sending AWS ML
request.", e);
Review Comment:
In general, error messages should not end with a `.` character. It would
also be helpful to rephrase the message along the following lines:
```suggestion
getLogger().error("Sending AWS ML Request failed", e);
```
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/ml/AwsMlProcessor.java:
##########
@@ -0,0 +1,148 @@
+package org.apache.nifi.processors.aws.ml;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.AmazonWebServiceRequest;
+import com.amazonaws.AmazonWebServiceResult;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentials;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processors.aws.AbstractAWSCredentialsProviderProcessor;
+
+public abstract class AwsMlProcessor<SERVICE extends AmazonWebServiceClient,
REQUEST extends AmazonWebServiceRequest, RESPONSE extends
AmazonWebServiceResult> extends
AbstractAWSCredentialsProviderProcessor<SERVICE> {
+ protected static final String AWS_TASK_ID_PROPERTY = "awsTaskId";
+ public static final PropertyDescriptor JSON_PAYLOAD = new
PropertyDescriptor.Builder()
+ .name("json-payload")
+ .displayName("JSON Payload")
+ .description("JSON Payload that represent an AWS ML Request. See
more details in AWS API documentation.")
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .build();
+ public static final PropertyDescriptor
MANDATORY_AWS_CREDENTIALS_PROVIDER_SERVICE =
+ new
PropertyDescriptor.Builder().fromPropertyDescriptor(AWS_CREDENTIALS_PROVIDER_SERVICE)
+ .required(true)
+ .build();
+ private static final List<PropertyDescriptor> PROPERTIES =
Collections.unmodifiableList(Arrays.asList(
+ JSON_PAYLOAD,
+ MANDATORY_AWS_CREDENTIALS_PROVIDER_SERVICE,
+ TIMEOUT,
+ SSL_CONTEXT_SERVICE,
+ ENDPOINT_OVERRIDE,
+ PROXY_CONFIGURATION_SERVICE,
+ PROXY_HOST,
+ PROXY_HOST_PORT,
+ PROXY_USERNAME,
+ PROXY_PASSWORD));
+ private final ObjectMapper mapper = JsonMapper.builder()
+ .configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true)
+ .build();
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+ RESPONSE response;
+ try {
+ response = sendRequest(buildRequest(session, context, flowFile),
context);
+ } catch (Exception e) {
+ session.transfer(flowFile, REL_FAILURE);
+ getLogger().error("Exception was thrown during sending AWS ML
request.", e);
+ return;
+ }
+
+ try {
+ writeToFlowFile(session, flowFile, response);
+ } catch (Exception e) {
+ session.transfer(flowFile, REL_FAILURE);
+ getLogger().error("Exception was thrown during writing aws
response to flow file.", e);
+ return;
+ }
+
+ try {
+ postProcessFlowFile(context, session, flowFile, response);
+ } catch (Exception e) {
+ session.transfer(flowFile, REL_FAILURE);
+ getLogger().error("Exception was thrown during AWS ML post
processing.", e);
+ return;
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ }
+
+ protected void postProcessFlowFile(ProcessContext context, ProcessSession
session, FlowFile flowFile, RESPONSE response) {
+ session.putAttribute(flowFile, AWS_TASK_ID_PROPERTY,
getAwsTaskId(context, response));
+ getLogger().debug("AWS ML task has been started with task id: {}",
getAwsTaskId(context, response));
+ }
+
+ protected REQUEST buildRequest(ProcessSession session, ProcessContext
context, FlowFile flowFile) throws JsonProcessingException {
+ REQUEST request;
+ try {
+ request = mapper.readValue(getPayload(session, context, flowFile),
+ getAwsRequestClass(context));
+ } catch (JsonProcessingException e) {
+ getLogger().error("Exception was thrown during AWS ML request
creation.", e);
+ throw e;
Review Comment:
In general, it is not necessary to log an error and then throw the
exception. If it is necessary to provide more context, it is better to throw
new exception with the specific error message.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/ml/AwsResponseMetadataDeserializer.java:
##########
@@ -0,0 +1,19 @@
+package org.apache.nifi.processors.aws.ml;
Review Comment:
This class is missing a license header.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/pom.xml:
##########
@@ -117,6 +117,26 @@
<version>1.19.0-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-translate</artifactId>
+ <version>1.12.328</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-polly</artifactId>
+ <version>1.12.328</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-transcribe</artifactId>
+ <version>1.12.328</version>
+ </dependency>
+ <dependency>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-textract</artifactId>
+ <version>1.12.328</version>
+ </dependency>
Review Comment:
Is there a reason for not using version 2 of the AWS SDK with the group ID
of `software.amazon.awssdk`?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
