You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "krisztina-zsihovszki (via GitHub)" <gi...@apache.org> on 2023/03/16 09:53:52 UTC
[GitHub] [nifi] krisztina-zsihovszki opened a new pull request, #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
krisztina-zsihovszki opened a new pull request, #7051:
URL: https://github.com/apache/nifi/pull/7051
<!-- 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. -->
# Summary
[NIFI-11255](https://issues.apache.org/jira/browse/NIFI-11255)
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [x] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [x] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [x] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
### Pull Request Formatting
- [x] Pull Request based on current revision of the `main` branch
- [x] Pull Request refers to a feature branch with one commit containing changes
# Verification
Please indicate the verification steps performed prior to pull request creation.
### Build
- [ ] Build completed using `mvn clean install -P contrib-check`
- [x] JDK 11
- [x] JDK 17
### Licensing
- [ ] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
- [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [ ] Documentation formatting appears as expected in rendered files
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1157058112
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AwsClientFactory.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.aws;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.Protocol;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.http.conn.ssl.SdkTLSSocketFactory;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import java.net.Proxy;
+import javax.net.ssl.SSLContext;
+import org.apache.http.conn.ssl.DefaultHostnameVerifier;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.ssl.SSLContextService;
+
+public class AwsClientFactory<ClientType extends AmazonWebServiceClient> {
+
+ private static final String DEFAULT_USER_AGENT = "NiFi";
+ private static final int MAXIMUM_CACHE_SIZE = 10;
+
+ private ComponentLog logger;
Review Comment:
The logger is not used, it may be removed.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AwsClientFactory.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.aws;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.Protocol;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.http.conn.ssl.SdkTLSSocketFactory;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import java.net.Proxy;
+import javax.net.ssl.SSLContext;
+import org.apache.http.conn.ssl.DefaultHostnameVerifier;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.ssl.SSLContextService;
+
+public class AwsClientFactory<ClientType extends AmazonWebServiceClient> {
+
+ private static final String DEFAULT_USER_AGENT = "NiFi";
+ private static final int MAXIMUM_CACHE_SIZE = 10;
+
+ private ComponentLog logger;
+
+ public AwsClientFactory(ComponentLog logger) {
+ this.logger = logger;
+ }
+
+ private final Cache<AwsClientDetails, ClientType> clientCache = Caffeine.newBuilder()
+ .maximumSize(MAXIMUM_CACHE_SIZE)
+ .build();
+
+ public ClientType getOrCreateClient(ProcessContext context, AWSCredentials credentials, AwsClientDetails clientDetails, AwsClientProvider provider) {
+ ClientConfiguration configuration = createConfiguration(clientDetails);
+ return clientCache.get(clientDetails, __ -> {
+ return (ClientType) provider.createClient(context, credentials, configuration);});
+ }
Review Comment:
The generic type for `AwsClientProvider` should be used. The lambda can be simplified too.
```suggestion
public ClientType getOrCreateClient(ProcessContext context, AWSCredentials credentials, AwsClientDetails clientDetails, AwsClientProvider<ClientType> provider) {
ClientConfiguration configuration = createConfiguration(clientDetails);
return clientCache.get(clientDetails, __ -> provider.createClient(context, credentials, configuration));
}
```
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AwsClientFactory.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.aws;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.Protocol;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.http.conn.ssl.SdkTLSSocketFactory;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import java.net.Proxy;
+import javax.net.ssl.SSLContext;
+import org.apache.http.conn.ssl.DefaultHostnameVerifier;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.ssl.SSLContextService;
+
+public class AwsClientFactory<ClientType extends AmazonWebServiceClient> {
+
+ private static final String DEFAULT_USER_AGENT = "NiFi";
+ private static final int MAXIMUM_CACHE_SIZE = 10;
+
+ private ComponentLog logger;
+
+ public AwsClientFactory(ComponentLog logger) {
+ this.logger = logger;
+ }
+
+ private final Cache<AwsClientDetails, ClientType> clientCache = Caffeine.newBuilder()
+ .maximumSize(MAXIMUM_CACHE_SIZE)
+ .build();
+
+ public ClientType getOrCreateClient(ProcessContext context, AWSCredentials credentials, AwsClientDetails clientDetails, AwsClientProvider provider) {
+ ClientConfiguration configuration = createConfiguration(clientDetails);
+ return clientCache.get(clientDetails, __ -> {
+ return (ClientType) provider.createClient(context, credentials, configuration);});
+ }
+
+ public ClientType getOrCreateClient(ProcessContext context, AWSCredentialsProvider credentialsProvider, AwsClientDetails configurationDetails, AwsClientProvider provider) {
Review Comment:
```suggestion
public ClientType getOrCreateClient(ProcessContext context, AWSCredentialsProvider credentialsProvider, AwsClientDetails clientDetails, AwsClientProvider provider) {
```
Also, the generic type + simplified lambda can be used here too.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1159528894
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java:
##########
@@ -382,18 +382,6 @@ private String parseRegionForVPCE(String url, String configuredRegion) {
}
}
- /**
- * Create client from the arguments
- * @param context process context
- * @param credentials static aws credentials
- * @param config aws client configuration
- * @return ClientType aws client
- *
- * @deprecated use {@link AbstractAWSCredentialsProviderProcessor#createClient(ProcessContext, AWSCredentialsProvider, ClientConfiguration)}
- */
- @Deprecated
- protected abstract ClientType createClient(final ProcessContext context, final AWSCredentials credentials, final ClientConfiguration config);
-
protected ClientType getClient() {
return client;
}
Review Comment:
The abstract processor should not store the client directly, only in the cache. It is quite confusing and error prone if some processors store the client in the field, others in the cache while everybody can access both type of client storage.
I would suggest making this `getClient()` method the entry point for the client cache (instead of `createClient()`). In this case it needs to be extended with `ProcessContext` parameter and also with `AwsClientDetails`.
Something like this:
```
protected ClientType getClient(final ProcessContext context, final AwsClientDetails clientDetails) {
return getAwsClientFactory().getOrCreateClient(context, clientDetails, this);
}
```
The processor implementation provides the `AwsClientDetails` object with the AWS region (e.g. parsed from FlowFile attributes in case of S3 processors)
The abstract class could also provide a convenience method for processors that do not have specific region handling:
```
protected ClientType getClient(final ProcessContext context) {
AwsClientDetails clientDetails = ... // create AwsClientDetails from the region in the context
return getClient(context, clientDetails);
}
```
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java:
##########
@@ -441,20 +431,27 @@ protected void setClientAndRegion(final ProcessContext context) {
* @param context The process context
* @return The created client
*/
- protected ClientType createClient(final ProcessContext context) {
- return createClient(context, getCredentials(context), createConfiguration(context));
+ public ClientType createClient(final ProcessContext context, AwsClientDetails awsClientDetails) {
+ AWSCredentials credentials = getCredentials(context);
+ final ClientType createdClient = getAwsClientFactory().getOrCreateClient(context, credentials, awsClientDetails, this);
+
+ setRegionAndInitializeEndpoint(awsClientDetails.getRegion(), context, createdClient);
+ return createdClient;
+ }
Review Comment:
As I mentioned at `getClient()`, I would move the cache access to that method and `createClient()` would only be responsible for creating the client (as before; the difference is that now we need `AwsClientDetails` parameter too).
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java:
##########
@@ -425,9 +413,11 @@ protected AWSCredentials getCredentials(final PropertyContext context) {
@OnShutdown
public void onShutdown() {
- if ( getClient() != null ) {
+ if (getClient() != null) {
getClient().shutdown();
}
+
+ getAwsClientFactory().getClients().forEach(AmazonWebServiceClient::shutdown);
Review Comment:
If we do not store the client separately, only the cached clients need to be shut down.
However, I noticed an old issue with this method: it is hooked on `@OnShutdown`. I believe the intent was `@OnStopped`. Now it runs only at NiFi shutdown which is mostly useless as the JVM will quit and client be destroyed. Also, it shuts down the lastly used client only (clients recreated at start/stop that are not shut down).
As it has not been an issue so far, I think we can simply delete this method for now.
In order to implement the client shutdown properly, we should also handle clients evicted from the cache too. It could be implemented in a separate jira.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AwsClientProvider.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.aws;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import org.apache.nifi.processor.ProcessContext;
+
+public interface AwsClientProvider<ClientType extends AmazonWebServiceClient> {
+ /**
+ * Create client from the arguments
+ * @param context process context
+ * @param credentials static AWS credentials
+ * @param config AWS client configuration
+ * @return ClientType AWS client
+ *
+ * @deprecated use {@link #createClient(ProcessContext, AWSCredentialsProvider, ClientConfiguration) }
+ */
+ @Deprecated
+ ClientType createClient(final ProcessContext context, final AWSCredentials credentials, final ClientConfiguration config);
+
+ /**
+ * Create AWS client using credentials provider. This is the preferred method for creating AWS clients
+ * @param context process context
+ * @param credentialsProvider AWS credentials provider
+ * @param config AWS client configuration
+ * @return ClientType AWS client
+ */
+ ClientType createClient(final ProcessContext context, final AWSCredentialsProvider credentialsProvider, final ClientConfiguration config);
Review Comment:
With moving the cache access logic to `getClient()`, this `AwsClientProvider` interface can expose `createClient(ProcessContext, AwsClientDetails)` only, which is the "original" / "main" method for creating the client. In this case we can get rid of `AWSCredentials`, `AWSCredentialsProvider` and `ClientConfiguration` in this interface (those are rather implementation details in the abstract processors).
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AwsClientFactory.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.aws;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.nifi.processor.ProcessContext;
+
+public class AwsClientFactory<ClientType extends AmazonWebServiceClient> {
Review Comment:
As `AwsClientProvider` creates the client and this class handles the cache, so I would rename it to `AwsClientCache`.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] asfgit closed pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "asfgit (via GitHub)" <gi...@apache.org>.
asfgit closed pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
URL: https://github.com/apache/nifi/pull/7051
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] Lehel44 commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "Lehel44 (via GitHub)" <gi...@apache.org>.
Lehel44 commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1142586898
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/s3/AbstractS3Processor.java:
##########
@@ -150,6 +160,16 @@ public abstract class AbstractS3Processor extends AbstractAWSCredentialsProvider
.dependsOn(SIGNER_OVERRIDE, CUSTOM_SIGNER)
.build();
+ public static final String S3_REGION_ATTRIBUTE = "s3.region" ;
+ static final AllowableValue ATTRIBUTE_DRIVEN_REGION = new AllowableValue("Use '" + S3_REGION_ATTRIBUTE + "' Attribute",
Review Comment:
What do you think of ATTRIBUTE_DEFINED_REGION? For the value the standard seems to be lowercase with dashes, so it would be "attribute-driven/defined-region". The description could be a bit simpler like "Uses the S3 Region Attribute as region"
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/s3/AbstractS3Processor.java:
##########
@@ -198,6 +218,88 @@ protected AmazonS3Client createClient(final ProcessContext context, final AWSCre
return s3;
}
+ protected Region resolveRegion(final ProcessContext context, final Map<String, String> attributes) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+
+ if (ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue)) {
+ regionValue = attributes.get(S3_REGION_ATTRIBUTE);
+ }
+
+ return parseRegionValue(regionValue);
+ }
+
+ private Region parseRegionValue(String regionValue) {
+ if (regionValue == null) {
+ throw new ProcessException(format("[%s] was selected as region source but [%s] attribute does not exist", ATTRIBUTE_DRIVEN_REGION, S3_REGION_ATTRIBUTE));
+ }
+
+ try {
+ return Region.getRegion(Regions.fromName(regionValue));
+ } catch (Exception e) {
+ throw new ProcessException(format("The [%s] attribute contains an invalid region value [%s]", S3_REGION_ATTRIBUTE, regionValue), e);
+ }
+ }
+
+ protected void setClientBasedOnRegionAttribute(final ProcessContext context, final Map<String, String> attributes) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+
+ if (ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue)) {
+ regionValue = attributes.get(S3_REGION_ATTRIBUTE);
+ Region region = parseRegionValue(regionValue);
Review Comment:
I think resolveRegion should be called here.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/s3/AbstractS3Processor.java:
##########
@@ -150,6 +160,16 @@ public abstract class AbstractS3Processor extends AbstractAWSCredentialsProvider
.dependsOn(SIGNER_OVERRIDE, CUSTOM_SIGNER)
.build();
+ public static final String S3_REGION_ATTRIBUTE = "s3.region" ;
+ static final AllowableValue ATTRIBUTE_DRIVEN_REGION = new AllowableValue("Use '" + S3_REGION_ATTRIBUTE + "' Attribute",
+ "Use '" + S3_REGION_ATTRIBUTE + "' Attribute",
+ "Inspects the '" + S3_REGION_ATTRIBUTE + "' FlowFile attribute to determine the used S3 region.");
+
+ public static final PropertyDescriptor S3_CUSTOM_REGION = new PropertyDescriptor.Builder()
Review Comment:
Should we call this 'custom' region? There is one option to define a custom region but overall it includes the regular regions from aws.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java:
##########
@@ -312,21 +312,9 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
*/
public abstract void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException;
- protected Region getRegionAndInitializeEndpoint(final ProcessContext context, final AmazonWebServiceClient client) {
- final Region region;
- // if the processor supports REGION, get the configured region.
- if (getSupportedPropertyDescriptors().contains(REGION)) {
- final String regionValue = context.getProperty(REGION).getValue();
- if (regionValue != null) {
- region = Region.getRegion(Regions.fromName(regionValue));
- if (client != null) {
- client.setRegion(region);
- }
- } else {
- region = null;
- }
- } else {
- region = null;
+ protected void initializeEndpoint(final Region region, final ProcessContext context, final AmazonWebServiceClient client) {
Review Comment:
I think the Region parameter is not needed, you can get it in the method by calling getRegion(context).
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] krisztina-zsihovszki commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "krisztina-zsihovszki (via GitHub)" <gi...@apache.org>.
krisztina-zsihovszki commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1143701000
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java:
##########
@@ -312,21 +312,9 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
*/
public abstract void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException;
- protected Region getRegionAndInitializeEndpoint(final ProcessContext context, final AmazonWebServiceClient client) {
- final Region region;
- // if the processor supports REGION, get the configured region.
- if (getSupportedPropertyDescriptors().contains(REGION)) {
- final String regionValue = context.getProperty(REGION).getValue();
- if (regionValue != null) {
- region = Region.getRegion(Regions.fromName(regionValue));
- if (client != null) {
- client.setRegion(region);
- }
- } else {
- region = null;
- }
- } else {
- region = null;
+ protected void initializeEndpoint(final Region region, final ProcessContext context, final AmazonWebServiceClient client) {
Review Comment:
Usually the region is taken from the context but in case it is is defined in the incoming flow file's 's3.region' attribute, it can not be fetched from the context.
This method is invoked by AbstractS3Processor specially, where the region can be taken from the flow file attribute.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1146356125
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/PutS3Object.java:
##########
@@ -503,11 +503,21 @@ private synchronized void ageoffLocalState(long ageCutoff) {
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) {
+
FlowFile flowFile = session.get();
if (flowFile == null) {
return;
}
+ try {
+ setClientAndRegion(context, flowFile.getAttributes());
+ } catch (Exception e) {
+ getLogger().error("Failed to initialize S3 client", e);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ return;
Review Comment:
@krisztina-zsihovszki Multiple threads can execute `onTrigger()` in parallel so it is not safe to create and store the client in the `client` field of the processor (`setClientAndRegion()` is doing that).
I suggest implementing client caching similar to [NIFI-10152](https://issues.apache.org/jira/browse/NIFI-10152) for Azure ADLS processors. It would also eliminate the recreation of the client in every onTrigger if the client can be reused (same region).
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] Lehel44 commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "Lehel44 (via GitHub)" <gi...@apache.org>.
Lehel44 commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1142676010
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/s3/AbstractS3Processor.java:
##########
@@ -198,6 +218,88 @@ protected AmazonS3Client createClient(final ProcessContext context, final AWSCre
return s3;
}
+ protected Region resolveRegion(final ProcessContext context, final Map<String, String> attributes) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+
+ if (ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue)) {
+ regionValue = attributes.get(S3_REGION_ATTRIBUTE);
+ }
+
+ return parseRegionValue(regionValue);
+ }
+
+ private Region parseRegionValue(String regionValue) {
+ if (regionValue == null) {
+ throw new ProcessException(format("[%s] was selected as region source but [%s] attribute does not exist", ATTRIBUTE_DRIVEN_REGION, S3_REGION_ATTRIBUTE));
+ }
+
+ try {
+ return Region.getRegion(Regions.fromName(regionValue));
+ } catch (Exception e) {
+ throw new ProcessException(format("The [%s] attribute contains an invalid region value [%s]", S3_REGION_ATTRIBUTE, regionValue), e);
+ }
+ }
+
+ protected void setClientBasedOnRegionAttribute(final ProcessContext context, final Map<String, String> attributes) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+
+ if (ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue)) {
+ regionValue = attributes.get(S3_REGION_ATTRIBUTE);
+ Region region = parseRegionValue(regionValue);
+
+ final AWSConfiguration awsConfiguration = getConfiguration(context, region);
+ this.client = awsConfiguration.getClient();
+ this.region = awsConfiguration.getRegion();
+ }
+ }
+
+ protected AWSConfiguration getConfiguration(final ProcessContext context, final Map<String, String> attributes) {
+ final Region region = resolveRegion(context, attributes);
+ return super.getConfiguration(context, region);
+ }
+
+ @Override
+ public List<ConfigVerificationResult> verify(final ProcessContext context, final ComponentLog verificationLogger, final Map<String, String> attributes) {
+ final List<ConfigVerificationResult> results = new ArrayList<>();
+
+ try {
+ getConfiguration(context, attributes);
+ results.add(new ConfigVerificationResult.Builder()
+ .outcome(Outcome.SUCCESSFUL)
+ .verificationStepName("Create S3 Client")
+ .explanation("Successfully created S3 Client")
+ .build());
+ } catch (final Exception e) {
+ verificationLogger.error("Failed to create S3 Client", e);
+ results.add(new ConfigVerificationResult.Builder()
+ .outcome(Outcome.FAILED)
+ .verificationStepName("Create S3 Client")
+ .explanation("Failed to crete S3 Client: " + e.getMessage())
+ .build());
+ }
+
+ return results;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ if (!isAttributeDrivenRegion(context)) {
+ setClientAndRegion(context);
+ }
+ }
+
+ private boolean isAttributeDrivenRegion(final ProcessContext context) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+ return ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue);
+ }
+
+ private static AllowableValue[] getCustomAvailableRegions() {
+ final List<AllowableValue> values = new ArrayList<>();
+ Collections.addAll(values, AbstractAWSCredentialsProviderProcessor.getAvailableRegions());
+ values.add(ATTRIBUTE_DRIVEN_REGION);
+ return values.toArray(new AllowableValue[0]);
+ }
Review Comment:
We can also use Apache Commons ArrayUtils::addAll
```suggestion
private static AllowableValue[] getCustomAvailableRegions() {
AllowableValue[] availableRegions = AbstractAWSCredentialsProviderProcessor.getAvailableRegions();
AllowableValue[] customRegions = new AllowableValue[]{ATTRIBUTE_DRIVEN_REGION};
return ArrayUtils.addAll(availableRegions, customRegions);
}
```
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] Lehel44 commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "Lehel44 (via GitHub)" <gi...@apache.org>.
Lehel44 commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1142673497
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/s3/AbstractS3Processor.java:
##########
@@ -198,6 +218,88 @@ protected AmazonS3Client createClient(final ProcessContext context, final AWSCre
return s3;
}
+ protected Region resolveRegion(final ProcessContext context, final Map<String, String> attributes) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+
+ if (ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue)) {
+ regionValue = attributes.get(S3_REGION_ATTRIBUTE);
+ }
+
+ return parseRegionValue(regionValue);
+ }
+
+ private Region parseRegionValue(String regionValue) {
+ if (regionValue == null) {
+ throw new ProcessException(format("[%s] was selected as region source but [%s] attribute does not exist", ATTRIBUTE_DRIVEN_REGION, S3_REGION_ATTRIBUTE));
+ }
+
+ try {
+ return Region.getRegion(Regions.fromName(regionValue));
+ } catch (Exception e) {
+ throw new ProcessException(format("The [%s] attribute contains an invalid region value [%s]", S3_REGION_ATTRIBUTE, regionValue), e);
+ }
+ }
+
+ protected void setClientBasedOnRegionAttribute(final ProcessContext context, final Map<String, String> attributes) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+
+ if (ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue)) {
+ regionValue = attributes.get(S3_REGION_ATTRIBUTE);
+ Region region = parseRegionValue(regionValue);
+
+ final AWSConfiguration awsConfiguration = getConfiguration(context, region);
+ this.client = awsConfiguration.getClient();
+ this.region = awsConfiguration.getRegion();
+ }
+ }
+
+ protected AWSConfiguration getConfiguration(final ProcessContext context, final Map<String, String> attributes) {
+ final Region region = resolveRegion(context, attributes);
+ return super.getConfiguration(context, region);
+ }
+
+ @Override
+ public List<ConfigVerificationResult> verify(final ProcessContext context, final ComponentLog verificationLogger, final Map<String, String> attributes) {
+ final List<ConfigVerificationResult> results = new ArrayList<>();
+
+ try {
+ getConfiguration(context, attributes);
+ results.add(new ConfigVerificationResult.Builder()
+ .outcome(Outcome.SUCCESSFUL)
+ .verificationStepName("Create S3 Client")
+ .explanation("Successfully created S3 Client")
+ .build());
+ } catch (final Exception e) {
+ verificationLogger.error("Failed to create S3 Client", e);
+ results.add(new ConfigVerificationResult.Builder()
+ .outcome(Outcome.FAILED)
+ .verificationStepName("Create S3 Client")
+ .explanation("Failed to crete S3 Client: " + e.getMessage())
+ .build());
+ }
+
+ return results;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ if (!isAttributeDrivenRegion(context)) {
+ setClientAndRegion(context);
+ }
+ }
+
+ private boolean isAttributeDrivenRegion(final ProcessContext context) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+ return ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue);
+ }
+
+ private static AllowableValue[] getCustomAvailableRegions() {
+ final List<AllowableValue> values = new ArrayList<>();
+ Collections.addAll(values, AbstractAWSCredentialsProviderProcessor.getAvailableRegions());
+ values.add(ATTRIBUTE_DRIVEN_REGION);
+ return values.toArray(new AllowableValue[0]);
+ }
Review Comment:
```suggestion
private static AllowableValue[] getCustomAvailableRegions() {
List<AllowableValue> values = new ArrayList<>(Arrays.asList(AbstractAWSCredentialsProviderProcessor.getAvailableRegions()));
values.add(ATTRIBUTE_DRIVEN_REGION);
return values.toArray(new AllowableValue[0]);
}
```
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] Lehel44 commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "Lehel44 (via GitHub)" <gi...@apache.org>.
Lehel44 commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1142672636
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/s3/AbstractS3Processor.java:
##########
@@ -198,6 +218,88 @@ protected AmazonS3Client createClient(final ProcessContext context, final AWSCre
return s3;
}
+ protected Region resolveRegion(final ProcessContext context, final Map<String, String> attributes) {
+ String regionValue = context.getProperty(S3_CUSTOM_REGION).getValue();
+
+ if (ATTRIBUTE_DRIVEN_REGION.getValue().equals(regionValue)) {
+ regionValue = attributes.get(S3_REGION_ATTRIBUTE);
+ }
+
+ return parseRegionValue(regionValue);
+ }
+
+ private Region parseRegionValue(String regionValue) {
+ if (regionValue == null) {
+ throw new ProcessException(format("[%s] was selected as region source but [%s] attribute does not exist", ATTRIBUTE_DRIVEN_REGION, S3_REGION_ATTRIBUTE));
+ }
+
+ try {
+ return Region.getRegion(Regions.fromName(regionValue));
+ } catch (Exception e) {
+ throw new ProcessException(format("The [%s] attribute contains an invalid region value [%s]", S3_REGION_ATTRIBUTE, regionValue), e);
+ }
+ }
+
+ protected void setClientBasedOnRegionAttribute(final ProcessContext context, final Map<String, String> attributes) {
Review Comment:
I recommend renaming this method to setClientAndRegion.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] turcsanyip commented on a diff in pull request #7051: NIFI-11255 Allowable value for 'Use s3.region Attribute'
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #7051:
URL: https://github.com/apache/nifi/pull/7051#discussion_r1169156291
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/test/java/org/apache/nifi/processors/aws/TestAwsClientCache.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.aws;
+
+import static com.amazonaws.regions.Regions.US_EAST_1;
+import static com.amazonaws.regions.Regions.US_WEST_2;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.amazonaws.regions.Region;
+import com.amazonaws.regions.Regions;
+import org.apache.nifi.processor.ProcessContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class TestAwsClientCache {
+
+ @Mock
+ private ProcessContext contextMock;
+ @Mock
+ private AwsClientProvider<AmazonWebServiceClient> awsClientProviderMock;
+ @Mock
+ private AmazonWebServiceClient awsClientMock1;
+ @Mock
+ private AmazonWebServiceClient awsClientMock2;
+ private AwsClientCache<AmazonWebServiceClient> clientCache;
+
+ @BeforeEach
+ public void setup() {
+ clientCache = new AwsClientCache<>();
+ }
+
+ @Test
+ public void testClientCreationNotNeededGetAwsTranslateJobStatusTest() {
+ final AwsClientDetails clientDetails = getClientDetails(US_WEST_2);
+ when(awsClientProviderMock.createClient(contextMock, clientDetails)).thenReturn(awsClientMock1);
+ final AmazonWebServiceClient client1 = clientCache.getOrCreateClient(contextMock, clientDetails, awsClientProviderMock);
+
+ final AwsClientDetails newClientDetails = getClientDetails(US_WEST_2);
+ final AmazonWebServiceClient client2 = clientCache.getOrCreateClient(contextMock, newClientDetails, awsClientProviderMock);
+ verify(awsClientProviderMock, times(1)).createClient(eq(contextMock), any(AwsClientDetails.class));
+ assertEquals(client1, client2);
Review Comment:
The two client objects should be the same by reference. So `assertSame()` could be used and it would describe the use case more explicitly..
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java:
##########
@@ -423,17 +408,28 @@ protected AWSCredentials getCredentials(final PropertyContext context) {
return new AnonymousAWSCredentials();
}
- @OnShutdown
- public void onShutdown() {
- if ( getClient() != null ) {
- getClient().shutdown();
- }
+ @OnStopped
+ public void onStopped() {
+ getAwsClientCache().clearCache();
+ }
+
+ /**
+ * Creates an AWS service client from the context or returns an existing client from the cache
+ * @param context The process context
+ * @param awsClientDetails details of the AWS client
+ * @return The created client
+ */
+ protected ClientType getClient(final ProcessContext context, AwsClientDetails awsClientDetails) {
+ return getAwsClientCache().getOrCreateClient(context, awsClientDetails, this);
}
- protected void setClientAndRegion(final ProcessContext context) {
- final AWSConfiguration awsConfiguration = getConfiguration(context);
- this.client = awsConfiguration.getClient();
- this.region = awsConfiguration.getRegion();
+ protected ClientType getClient(final ProcessContext context) {
+ final AwsClientDetails awsClientDetails = new AwsClientDetails(getRegion(context));
+ return getClient(context, awsClientDetails);
+ }
+
+ protected AwsClientCache<ClientType> getAwsClientCache() {
+ return awsClientCache;
Review Comment:
`getAwsClientCache()` is not called or overridden in child classes. The method should be `private` or even the field could be accessed directly instead of the getter method.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AbstractAWSProcessor.java:
##########
@@ -446,15 +442,26 @@ protected ClientType createClient(final ProcessContext context) {
}
/**
- * Parses and configures the client and region from the context.
+ * Create client from the arguments
+ * @param context process context
+ * @param credentials static aws credentials
+ * @param config aws client configuration
+ * @return ClientType aws client
+ *
+ * @deprecated use {@link AbstractAWSCredentialsProviderProcessor#createClient(ProcessContext, AWSCredentialsProvider, ClientConfiguration)}
+ */
+ @Deprecated
+ protected abstract ClientType createClient(final ProcessContext context, final AWSCredentials credentials, final ClientConfiguration config);
+
+ /**
+ * Parses and configures the client from the context.
* @param context The process context
* @return The parsed configuration
*/
protected AWSConfiguration getConfiguration(final ProcessContext context) {
- final ClientType client = createClient(context);
- final Region region = getRegionAndInitializeEndpoint(context, client);
-
- return new AWSConfiguration(client, region);
+ final AwsClientDetails awsClientDetails = new AwsClientDetails(getRegion(context));
+ final ClientType createdClient = getClient(context, awsClientDetails);
+ return new AWSConfiguration(createdClient, awsClientDetails.getRegion());
Review Comment:
I'd suggest removing this method and the `AWSConfiguration` inner class.
`onScheduled()` methods can call `getClient(ProcessContext)` directly which is more explicit and readable I believe.
`verify()` methods should not call `getClient()` (directly or indirectly via `getConfiguration()`) because `getClient()` populates the cache with the created client. If the user modifies some properties after verification, the old client gets stuck in the cache and no new client is created with the changed property values. `verify()` methods can call `createClient(ProcessContext)` which does not access / initialize the cache.
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-abstract-processors/src/main/java/org/apache/nifi/processors/aws/AwsClientCache.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * 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.aws;
+
+import com.amazonaws.AmazonWebServiceClient;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import org.apache.nifi.processor.ProcessContext;
+
+public class AwsClientCache<ClientType extends AmazonWebServiceClient> {
+
+ private static final int MAXIMUM_CACHE_SIZE = 10;
+
+ private final Cache<AwsClientDetails, ClientType> clientCache = Caffeine.newBuilder()
+ .maximumSize(MAXIMUM_CACHE_SIZE)
+ .build();
+
+ public ClientType getOrCreateClient(final ProcessContext context, final AwsClientDetails clientDetails, final AwsClientProvider<ClientType> provider) {
+ return clientCache.get(clientDetails, ignored -> provider.createClient(context, clientDetails));
+ }
+
+ public void clearCache() {
+ clientCache.cleanUp();
Review Comment:
`Cache.invalidateAll()` needs to be called in order to remove the items and to clear the cache.
`cleanUp()` just _"performs any pending maintenance operations"_.
Not sure if `cleanUp()` is needed after `invalidateAll()`. For sure, we can use both.
A unit test may be added to check the proper cleaning.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org