You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "markap14 (via GitHub)" <gi...@apache.org> on 2026/03/10 19:40:55 UTC
[PR] NIFI-15258: Initial implementation of Connectors [nifi]
markap14 opened a new pull request, #10993:
URL: https://github.com/apache/nifi/pull/10993
NIFI-15258: Fixed vulnerable lz4 dependency
NIFI-15258: Fixed checkstyle violations/pmd
NIFI-15258: Addressed PMD findings
NIFI-15259: Added REST API for Connectors
Introduced comprehensive REST endpoints for managing Connectors through the NiFi web interface and API clients.
Key capabilities:
- Full lifecycle management (create, read, update, delete)
- Start/stop control with run status endpoints
- Multi-step configuration with property groups supporting string literals, asset references, and secret references
- Asynchronous configuration verification with polling support
- Access to connector's managed flow and process group status
- Search functionality within connector's encapsulated flow
Cluster support:
- Request replication across cluster nodes
- Response merging for dynamic property allowable values that may differ between nodes
- Consistent configuration step and property descriptor handling
Infrastructure:
- Audit logging for all connector configuration changes
- Authorization integration with NiFi's policy framework
- UI extension support for connector-specific interfaces
- Frontend updates for connector access policy management
NIFI-15323: Adding operate permissions to the ConnectorEntity. (#10625)
This closes #10625
NIFI-15322: Require all property descriptors within a PropertyGroup / ConfigurationStep are unique (#10626)
* NIFI-15322: Huge refactoring of Connectors to require that <Step Name, Property Name> tuple is unique. Requiring that is simple, but now that we enforce that, we can greatly simplify much of the code.
* NIFI-15322: Added StepConfiguration to simplify API rather than using Map<String, ConnectorValueReference> in some places
* NIFI-15322: Now that we enforce unique property names per step, introduced StepConfigurationContext to further reduce API complexity
NIFI-15326: Adding support of configuration step documentation. (#10631)
NIFI-15322: Adapting to new connector configuration step model.
NIFI-15330: Allow GhostConnector to be updated (#10632)
NIFI-15312: Implementation of Parameter Provider based Secrets Manager; updated mock framework to support secrets; validation cleanup (#10638)
NIFI-15336: Created AuthorizableSecret and implemented ParameterProviderSecretProvider to provide the Parameter Provider as the associated Authorizable (#10644)
NIFI-15258: Enable Custom UI to retrieve connector with flow contexts (#10652)
NIFI-15352: Added ConnectorClient to toolkit-cli so that it can be us… (#10654)
* NIFI-15352: Added ConnectorClient to toolkit-cli so that it can be used in system tests; created system tests for Connectors; bug fixes found running tests; updated Secret/SecretReference to support fully qualified names
* NIFI-15352: Addressed review feedback
NIFI-15343: Adding an endpoint to return available secrets to the con… (#10645)
* NIFI-15343: Adding an endpoint to return available secrets to the connector configuration wizard.
* NIFI-15343: Adding providerId to SecretDTO.
* NIFI-15343: Skipping ParameterProviders that are invalid or validating when fetching secrets.
* NIFI-15343: Adding fully qualified name to SecretDTO.
- Fixing rebase issue.
This closes #10645
NIFI-15315 Add support for assets in connectors (#10647)
- Support multiple asset references
- Clean up of assets in StandardConnectorDAO after apply and discard
- Add endpoint for listing assets for a given connector
- Add endpoint for discarding the working config of a connector
- Add endpoint for retrieving asset content
- Add system test for connectors that uploads an asset
* Fix issues after rebasing, enable ClusteredConnectorAssetsIT
NIFI-15361: Allowing configuration step documentation to be returned … (#10667)
* NIFI-15361: Allowing configuration step documentation to be returned independent of any connector instances.
NIFI-15369: Allow ConfigurationStep to depend on another (ConfigurationStep,Property) tuple (#10669)
NIFI-15367: Ensure that Connectors' implicit parameter contexts are n… (#10665)
* NIFI-15367: Ensure that Connectors' implicit parameter contexts are not registered with ParameterContextManager; added system test to verify; fixed bug in ConnectorAssetsIT where it did not wait for connector to return to its 'STOPPED' state before attempting to delete
* NIFI-15367: Bug fixes; review feedback
NIFI-15370: Add Connector ID to process groups and do not register create process groups if they have a connector ID. (#10673)
NIFI-15356: Adding authorization to the StandardNiFiConnectorWebContext. (#10660)
NIFI-15353: Adding support for rendering Connector Documentation. (#10683)
* NIFI-15353: Adding support for rendering Connector Documentation.
* NIFI-15353: Addressing review feedback.
* NIFI-15353: Using mergeMap instead of switchMap to allow for concurrent doc loading.
This closes #10683
NIFI-15429: Adding an optional query parameter for specifying which p… (#10732)
* NIFI-15429: Adding an optional query parameter for specifying which process group should be returned within the Connectors flow.
* NIFI-15429: Including managed process group id in the ConnectorDTO.
* NIFI-15429: Addressing review feedback.
* NIFI-15429: Updating path for fetching a Connector flow to require the desired Process Group ID.
* NIFI-15429: Fixing failing unit test.
NIFI-15430: Ensure that if we fail to initialize a Connector, we crea… (#10733)
* NIFI-15430: Ensure that if we fail to initialize a Connector, we create a GhostConnector instead and ensure that we also proivde the extensionMissing flag on ConnectorNode
* NIFI-15430: Added extensionMissing flag to Connector DTO
* NIFI-15430: If unable to load initial flow of a Connector, make ghosted connector instead
NIFI-15427: Added abiliy to drop flowfiles / drain flowfiles from a C… (#10730)
* NIFI-15427: Added abiliy to drop flowfiles / drain flowfiles from a Connector; added some system tests to verify existing behavior; fixed existing issue with IT
* NIFI-15427: Fixed Exception message in case unable to purge FlowFiles due to Connector state
* NIFI-15427: Ensured thread safety of state transitions when draining/purging FlowFiles
* NIFI-15427: Addressed review feedback
NIFI-15433: If connector validation throws an Exception keep trying u… (#10736)
* NIFI-15433: If connector validation throws an Exception keep trying until it completes (with a timeout between retries); some bug fixes
* NIFI-15433: Addressed review feedback
* NIFI-15433: Fixed failing unit tests
NIFI-15434: Ensure that we start or stop connectors on flow sync (#10737)
NIFI-15445: Removed 'disabled' state from Connectors; also fixed bug … (#10749)
* NIFI-15445: Removed 'disabled' state from Connectors; also fixed bug to ensure that if we fail to apply an update to a Connector that we stop it.
* NIFI-15445: Addressed review feedback
* NIFI-15445: Fixed checkstyle
NIFI-15440: Implementation of ConnectorActions (#10748)
* NIFI-15440: Implementation of ConnectorActions
* NIFI-15440: Addressed review feedback
* NIFI-15440: Addressed issues that occurred after rebase
* NIFI-15440: Addressed review feedback
NIFI-15439: Ensure that Process Groups are accessible from the approp… (#10742)
* NIFI-15439: Ensure that Process Groups are accessible from the appropriate endpoint but not accessible elsewhere
* NIFI-15439: FlowAnalyzingRegistryClient should use an explicit Connector ID of null when retrieving Process Group from FlowManager
NIFI-15468: Ensure that we properly initialize ConnectorNode even if a Ghost Connector is created (#10775)
This closes #10775
NIFI-15446: When invoking ConnectorMethod, make sure to serialize/des… (#10750)
* NIFI-15446: When invoking ConnectorMethod, make sure to serialize/deserialize the arguments and return value
* NIFI-15446: Addressed review feedback
NIFI-15376: Adding dependencies to the configuration step dto. (#10674)
NIFI-15453: Adding an endpoint to return all controller services for … (#10770)
* NIFI-15453: Adding an endpoint to return all controller services for a process group in a connector.
* NIFI-15453: Addressing review feedback.
This closes #10770
NIFI-15472: Added addAsset methods to ConnectorTestRunner (#10776)
NIFI-15482: MockConnectorAssetManager requires SPI file to start (#10783)
NIFI-15451: Added ability for Connectors to retrieve bundles availabl… (#10756)
* NIFI-15451: Added ability for Connectors to retrieve bundles available for component types and updated VersionedFlowUtils to make use of it to easily update Versioned components
* NIFI-15451: Added additional unit tests to VersionedFlowUtils; updates to how we compare component versions
* NIFI-15451: Fixed PMD violation
NIFI-15481: Updated ConnectorActions to reflect that starting is not … (#10782)
* NIFI-15481: Updated ConnectorActions to reflect that starting is not allowed when invalid. Also updated ValidationState to ensure that we don't return any ValidationResult with a valid flag of true as a 'validation error'.
* NIFI-15481: Addressed review feedback
NIFI-15258: Fixes necessary after rebasing against main
NIFI-15258: Addressed checkstyle violations
NIFI-15461: Added ability to initiate drainage of Connector's FlowFiles and calce; added tests to verify; some bug fixes (#10767)
NIFI-15489: Addressed the 4 bugs called out in Jira: Wrong delimiter in StandardConnectorPropertyValue.asList, ConnectorValidationContextBridge.getProperty() does not resolve property value properly; StandaloneProcessGroupLifecycle.stop() does not await controller service disabling; switched hashmap in StandaloneProcessGroupLifecycle to ConcurrentHashMap (#10798)
- Addressed another similar bug in which an invalid connector that is expected to be running would prevent NiFi from starting
NIFI-15490: If any processors or controller services are missing, make Connector invalid (#10799)
NIFI-15485: Include provider name when providing fqn for secrets (#10801)
NIFI-15497: Fixed unit tests (#10805)
NIFI-15258: Fixes for system tests that needed to be updated from 2.7.0-SNAPSHOT to 2.8.0-SNAPSHOT (#10807)
NIFI-15495 Restart Connectors that reference assets that were synchro… (#10806)
* NIFI-15495 Restart Connectors that reference assets that were synchronized
- Ensure Connectors re-resolve property values before starting
- Ensure asset clean up happens only after applyUpdate fully finishes
- Add connector asset properties to default nifi.properties
* Fix system test
* Encapsulate restart logic in new method on ConnectorNode
* Fix JavaDoc
NIFI-15465 Allow client-specified connector component ids (#10768)
Add verifyCreate to ConnectorDAO and test cases
NIFI-15467: Ensure that Connectors' versions are automatically resolv… (#10811)
* NIFI-15467: Ensure that Connectors' versions are automatically resolved on startup
- Allow connectors to specify Bundle Compatability when updating flow
- Default to BundleCompatability.RESOLVE_BUNDLE when connectors are updating flows
- Bug fixes; updated MockConnectorInitializationContext to extend from StandardConnectorInitializationContext to cut down on code duplication
* NIFI-15467: Addressed review feedback
NIFI-15529: Fix to StatelessGroupLifecycle to ensure that we set the desired state of Stateless Group Node before attempting to schedule; updated StandardProcessScheduler.startStatelessGroup to make it more clear if this is left out again at some point (#10832)
NIFI-15525: Ensure that we call onStepConfigured when we restore flow; ensure that we synchronized Parmaeter Providers before Connectors (#10835)
NIFI-15509: Enable Component reload additional classpath resources if needed (#10812)
NIFI-15536: Change Connector API to use DescribedValue vs AllowableValue (#10841)
NIFI-15511: Added endpoints for purging all FlowFiles for a given Con… (#10833)
* NIFI-15511: Added endpoints for purging all FlowFiles for a given Connector; added method to ConnectorClient to call these endpoints; updated system tests to use these endpoints when tearing down flows; some bug fixes
* NIFI-15511: Added WRITE permission requirement for retrieving and deleting a Connector Purge request
NIFI-15514: Ensure that Parameter Contexts are assigned to all Proces… (#10815)
* NIFI-15514: Ensure that Parameter Contexts are assigned to all Process Groups in a Connector and not just the top-level group. Ensure that all components are started when Connector starts instead of just Processors and Controller Services
- When Working Context is recreated, ensure that we appropriately apply Parameter Context to newly created Process Group(s)
- Ensure that when we cleanup unused assets for Connectors that we consider any assets that are referenced in either the Working or Active context instead of just the Active context
- Ensure that when we stop Process Group we call all tasks in background threads instead of calling .thenRun which could potentially run in the foreground thread
* NIFI-15514: Update parameter context assignment to occur during sync.
* NIFI-15514: Fixes around ensuring that processors/controller services are properly configured and notified of any configuration changes when parameters change
- Removed the updateParameterContexts from ProcessGroup.updateFlow, which was added in a previous commit as we went a different direction for the fix
---------
NIFI-15538: When starting/stopping components allow specifying whethe… (#10843)
* NIFI-15538: When starting/stopping components allow specifying whether or not the action should be recursive; code cleanup and simplification to use a Virtual Thread to execute code sequentially instead of chaining CompletableFutures.
* NIFI-15538: Addressed review feedback
---------
NIFI-15480 Added ability to drop FlowFiles selectively using a Predicate (#10848)
* NIFI-15480: Added ability to drop FlowFiles selectively using a Predicate Added system tests to verify selective dropping of flowfiles from a Connector. Also made significant updates so that diagnostic bundles are captured when a system test fails and diagnostic bundle also contains details of each Connection's FlowFileQueue. This was necessary in aiding the debugging of the new feature but is also something that has been missing for some time.
* NIFI-15480: Ensure that when we perform a selective drop on a FlowFileQueue that we pass appropriate SWAP_FILE_DELETED / SWAP_FILE_RENAMED events to the FlowFile Repository
NIFI-15549: Adding endpoints for viewing and clearing state for compo… (#10855)
* NIFI-15549: Adding endpoints for viewing and clearing state for components that are managed by a Connector.
* NIFI-15549: Addressing review feedback.
NIFI-15557: Allow Connectors to enable a Controller Service using overridden property values (#10862)
NIFI-15488: Added significant number of debug log messages as well as some info/error for connector-related events (#10803)
NIFI-15544: If a Connector requires processors or controller services that are unavailable, make Connector invalid but not ghosted (#10851)
NIFI-15558: Sensitive Parameter References should resolve to value. (#10863)
NIFI-15539: Updating the actions in FlowFileQueueResource to support … (#10846)
* NIFI-15539: Updating the actions in FlowFileQueueResource to support interacting with queues in flows managed by a Connector.
* NIFI-15539: Updating resource endpoint to include data policies for connectors.
NIFI-15565: Code cleanup; improvements to .cursorrules to provide bet… (#10869)
* NIFI-15565: Code cleanup; improvements to .cursorrules to provide better instructions regarding automated tests
* NIFI-15565: Broke apart cursorrules file into the new format of .cursor/rules/*
NIFI-15355 Connector Configuration Repository (#10876)
* NIFI-15255 Introduce ConnectorConfigurationRepository Extension Interface
* Add new property key to default nifi.properties file
* Address peer review comments
NIFI-15559: Enforce data permissions when purging data from a Connector. (#10867)
This closes #10867
NIFI-15258: Fixes as a result of rebasing against main
NIFI-15577: Including connector id in provenance event dto. (#10883)
* NIFI-15577: Including connector id in provenance event dto.
- Fixing go to routing logic in the provenance event table.
* NIFI-15577: Fixing back navigation context when navigating to provenance from the flow designer.
This closes #10883
NIFI-15581: Add support for ControllerService Mocks for Connectors. (#10885)
NIFI-15579: When synchronizing PG when Versioned flow for stateless group node, do not synchronize parameter contexts if group is within a Connector (#10884)
NIFI-15605 - Rebase Connectors branch with main for 2.9.0-SNAPSHOT
NIFI-15606: Connector Mock War implementation. (#10907)
* NIFI-15606: Connector Mock War implementation.
* Load war files from Connector Nar
* Mock Connector Web Context
* Allow test runner to set http port
* NIFI-15606: Connector Mock War implementation.
* Code Review Feedback
NIFI-15611: Fetchable Values should pull from current FlowContext (#10906)
NIFI-15618: Bug fixes around handling of FlowFileActivity with child groups and stateless groups (#10912)
NIFI-15598: When enabling referenced controller services in a connector, skip references from properties with unsatisfied dependencies (#10897)
NIFI-15615: Ensure that we include necessary values in the ConnectorStatusDTO (#10913)
NIFI-15610 Add asset management to ConnectorConfigurationProvider (#10909)
NIFI-15621: Allow Configuration to Validate with ValidationContext (#10916)
* Currently this is only implemented for processors in schedule periods
* Allows connector to pass Validation Context to be used instead of current parameter context.
NIFI-15640: NPE in the applyFlow should NOT crash the runtime. (#10931)
* NIFI-15640: NPE in the applyFlow should NOT crash the runtime.
* Leave the flow in a state of UPDATE_FAILED
* With validationState of INVALID.
* NIFI-15640: NPE in the applyFlow should NOT crash the runtime.
* Code Review Feedback
NIFI-15629 Add ConnectorRequestContext (#10924)
NIFI-15663: Update to use new ComponentHierarchyScope instead of on facade/lifecycle classes (#10958)
NIFI-15680: Not including global parameter contexts and parameters in connector search results. (#10980)
This closes #10980
NIFI-15604: Expose VersionedExternalFlow that represents the Active/Working flow from ConnectorTestRunner (#10901)
NIFI-15648 - Resolve SECRET property values before fetchAllowableValues and verify in working flow context (#10937)
<!-- 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-00000](https://issues.apache.org/jira/browse/NIFI-00000)
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [ ] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [ ] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [ ] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
- [ ] Pull request contains [commits signed](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits) with a registered key indicating `Verified` status
### Pull Request Formatting
- [ ] Pull Request based on current revision of the `main` branch
- [ ] 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 `./mvnw clean install -P contrib-check`
- [ ] JDK 21
- [ ] JDK 25
### 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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2984750200
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockExtensionDiscoveringManager.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.bundle.Bundle;
+import org.apache.nifi.bundle.BundleCoordinate;
+import org.apache.nifi.bundle.BundleDetails;
+import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.nar.InstanceClassLoader;
+import org.apache.nifi.nar.StandardExtensionDiscoveringManager;
+import org.apache.nifi.processor.Processor;
+
+import java.io.File;
+import java.net.URL;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class MockExtensionDiscoveringManager extends StandardExtensionDiscoveringManager {
+ private final ConcurrentMap<String, ClassLoader> mockComponentClassLoaders = new ConcurrentHashMap<>();
+
+ public synchronized void addProcessor(final Class<? extends Processor> mockProcessorClass) {
+ final BundleDetails bundleDetails = new BundleDetails.Builder()
+ .workingDir(new File("target/work/extensions/mock-bundle"))
Review Comment:
I'm not sure it really matters - I don't believe we actually write anything to that directory or anything like that. It just has to be set to *something* or else an Exception will get thrown by the parent class.
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2984740057
##########
nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.components.DescribedValue;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.connector.AssetReference;
+import org.apache.nifi.components.connector.ConnectorValueReference;
+import org.apache.nifi.components.connector.FlowUpdateException;
+import org.apache.nifi.components.connector.SecretReference;
+import org.apache.nifi.components.connector.StepConfiguration;
+import org.apache.nifi.flow.VersionedExternalFlow;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+public interface ConnectorTestRunner extends Closeable {
Review Comment:
Yeah, great call. I intended to do that before pushing the PR but it slipped my mind. Will update.
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2991244281
##########
nifi-commons/nifi-connector-utils/pom.xml:
##########
@@ -0,0 +1,42 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <!--
+ 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.
+ -->
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-commons</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-connector-utils</artifactId>
Review Comment:
Thanks, that sounds reasonable given expected usage.
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on PR #10993:
URL: https://github.com/apache/nifi/pull/10993#issuecomment-4135124089
Great catch @exceptionfactory - easy fix, we actually have a flag on the specific method for whether or not to log that, so super easy fix. Updated.
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2991240028
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockExtensionDiscoveringManager.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.bundle.Bundle;
+import org.apache.nifi.bundle.BundleCoordinate;
+import org.apache.nifi.bundle.BundleDetails;
+import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.nar.InstanceClassLoader;
+import org.apache.nifi.nar.StandardExtensionDiscoveringManager;
+import org.apache.nifi.processor.Processor;
+
+import java.io.File;
+import java.net.URL;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class MockExtensionDiscoveringManager extends StandardExtensionDiscoveringManager {
+ private final ConcurrentMap<String, ClassLoader> mockComponentClassLoaders = new ConcurrentHashMap<>();
+
+ public synchronized void addProcessor(final Class<? extends Processor> mockProcessorClass) {
+ final BundleDetails bundleDetails = new BundleDetails.Builder()
+ .workingDir(new File("target/work/extensions/mock-bundle"))
Review Comment:
Thanks, given the usage described, that sounds acceptable.
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory merged PR #10993:
URL: https://github.com/apache/nifi/pull/10993
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2982638665
##########
nifi-commons/nifi-connector-utils/pom.xml:
##########
@@ -0,0 +1,42 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <!--
+ 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.
+ -->
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-commons</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-connector-utils</artifactId>
Review Comment:
I'd be okay with `nifi-connector-common`. But I would not consider it an API level contract. It is merely a util class that makes it easier to manipulate or find components within a `VersionedExternalFlow`.
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2978886840
##########
nifi-connectors/pom.xml:
##########
@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-connectors</artifactId>
+ <packaging>pom</packaging>
+
+ <modules>
+ <module>nifi-kafka-to-s3-bundle</module>
+ </modules>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-utils</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
Review Comment:
Is this necessary? It seems like setting the dependency version here is duplicative.
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockAuditService.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.action.Action;
+import org.apache.nifi.admin.service.AuditService;
+import org.apache.nifi.history.History;
+import org.apache.nifi.history.HistoryQuery;
+import org.apache.nifi.history.PreviousValue;
+
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+public class MockAuditService implements AuditService {
Review Comment:
Does this need to be public or can it be package-private?
##########
nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/pom.xml:
##########
@@ -0,0 +1,238 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-kafka-to-s3-bundle</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-kafka-to-s3-integration-tests</artifactId>
+
+ <repositories>
+ <repository>
+ <id>confluent</id>
+ <url>https://packages.confluent.io/maven/</url>
+ </repository>
+ </repositories>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock-api</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+
+ <!-- Test Containers for Kafka and S3 -->
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers-kafka</artifactId>
+ <version>2.0.1</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>localstack</artifactId>
+ <version>1.20.4</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Kafka client for producing test data -->
+ <!-- Override vulnerable kafka clients -->
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>4.1.1</version>
Review Comment:
Given the relationship to the Kafka Processors, it seems like the Kafka version should be promoted to the root Maven configuration and shared with the Kafka component bundle to avoid mismatches.
##########
nifi-commons/nifi-connector-utils/pom.xml:
##########
@@ -0,0 +1,42 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <!--
+ 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.
+ -->
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-commons</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-connector-utils</artifactId>
Review Comment:
Although some other modules in `nifi-common` have the `utils` suffix, what about `nifi-connector-shared` or `nifi-connector-common`, or `nifi-connector-support`? It looks like the utilities class supports construction of Connectors, so it is also to an API level contract, thus something a bit more concrete than `utils` seems better.
##########
nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.components.DescribedValue;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.connector.AssetReference;
+import org.apache.nifi.components.connector.ConnectorValueReference;
+import org.apache.nifi.components.connector.FlowUpdateException;
+import org.apache.nifi.components.connector.SecretReference;
+import org.apache.nifi.components.connector.StepConfiguration;
+import org.apache.nifi.flow.VersionedExternalFlow;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+public interface ConnectorTestRunner extends Closeable {
Review Comment:
This seems like a central interface for testing purposes, so it would be helpful to expand the documentation to cover most of the methods.
##########
nifi-connectors/nifi-kafka-to-s3-bundle/pom.xml:
##########
@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connectors</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-kafka-to-s3-bundle</artifactId>
+ <packaging>pom</packaging>
+
+ <modules>
+ <module>nifi-kafka-to-s3-connector</module>
+ <module>nifi-kafka-to-s3-nar</module>
+ <module>nifi-kafka-to-s3-integration-tests</module>
+ </modules>
+
+ <dependencyManagement>
+ <dependencies>
+ </dependencies>
+ </dependencyManagement>
Review Comment:
This empty block can be removed.
##########
nifi-connector-mock-bundle/nifi-connector-mock-server-nar/pom.xml:
##########
@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock-bundle</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-connector-mock-server-nar</artifactId>
+ <packaging>nar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock-server</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-framework-nar</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
+ </dependencies>
+
+ <repositories>
+ <!-- Shibboleth Repository required for OpenSAML -->
+ <!-- This is required because nifi-framework-nar is the parent/NAR Dependency, andi t requires shibboleth. -->
Review Comment:
```suggestion
<!-- Required for nifi-framework-nar dependencies -->
```
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockExtensionMapper.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.flow.Bundle;
+import org.apache.nifi.flow.VersionedControllerService;
+import org.apache.nifi.flow.VersionedProcessor;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class MockExtensionMapper {
+
+ private final Map<String, String> processorMocks = new HashMap<>();
+ private final Map<String, String> controllerServiceMocks = new HashMap<>();
+
+ public void mockProcessor(final String processorType, final String mockProcessorClassName) {
+ processorMocks.put(processorType, mockProcessorClassName);
+ }
+
+ public void mockControllerService(final String controllerServiceType, final String mockControllerServiceClassName) {
+ controllerServiceMocks.put(controllerServiceType, mockControllerServiceClassName);
+ }
+
+ public void mapProcessor(final VersionedProcessor processor) {
+ final String type = processor.getType();
+ final String implementationClassName = processorMocks.get(type);
+ if (implementationClassName == null) {
+ return;
+ }
+
+ processor.setType(implementationClassName);
+ processor.setBundle(new Bundle("org.apache.nifi.mock", implementationClassName, "1.0.0"));
Review Comment:
It looks like the bundle package and version should be declared as static variables.
##########
nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/HashMapSnapshot.java:
##########
@@ -175,14 +175,18 @@ public void update(final Collection<T> records) {
switch (updateType) {
case DELETE:
- recordMap.remove(recordId);
+ if (recordId != null) {
Review Comment:
Are these changes required for Connectors? It seems like they could be handled independently.
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorAssetManager.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.asset.Asset;
+import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.asset.AssetManagerInitializationContext;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A Mock implementation of AssetManager for use in ConnectorTestRunner tests.
+ * This implementation stores assets in a temporary directory structure.
+ */
+public class MockConnectorAssetManager implements AssetManager {
+
+ private static final String ASSET_STORAGE_LOCATION_PROPERTY = "directory";
+ private static final String DEFAULT_ASSET_STORAGE_LOCATION = "target/mock_connector_assets";
+
+ private final Map<String, Asset> assets = new ConcurrentHashMap<>();
+ private volatile File assetStorageLocation;
+
+ @Override
+ public void initialize(final AssetManagerInitializationContext context) {
+ final String storageLocation = context.getProperties().getOrDefault(ASSET_STORAGE_LOCATION_PROPERTY, DEFAULT_ASSET_STORAGE_LOCATION);
+ assetStorageLocation = new File(storageLocation);
+
+ if (!assetStorageLocation.exists()) {
+ try {
+ Files.createDirectories(assetStorageLocation.toPath());
+ } catch (final IOException e) {
+ throw new RuntimeException("Failed to create asset storage directory: " + storageLocation, e);
+ }
+ }
+ }
+
+ @Override
+ public Asset createAsset(final String ownerId, final String assetName, final InputStream contents) throws IOException {
+ final String assetId = UUID.randomUUID().toString();
+ return saveAsset(ownerId, assetId, assetName, contents);
+ }
+
+ @Override
+ public Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents) throws IOException {
+ final File assetFile = getFile(ownerId, assetId, assetName);
+ final File parentDir = assetFile.getParentFile();
+
+ if (!parentDir.exists()) {
+ Files.createDirectories(parentDir.toPath());
+ }
+
+ Files.copy(contents, assetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
+
+ final Asset asset = new MockAsset(assetId, ownerId, assetName, assetFile, null);
+ assets.put(assetId, asset);
+ return asset;
+ }
+
+ @Override
+ public Optional<Asset> getAsset(final String id) {
+ return Optional.ofNullable(assets.get(id));
+ }
+
+ @Override
+ public List<Asset> getAssets(final String ownerId) {
+ final List<Asset> ownerAssets = new ArrayList<>();
+ for (final Asset asset : assets.values()) {
+ if (asset.getOwnerIdentifier().equals(ownerId)) {
+ ownerAssets.add(asset);
+ }
+ }
+ return ownerAssets;
+ }
+
+ @Override
+ public Asset createMissingAsset(final String ownerId, final String assetName) {
+ final String assetId = UUID.randomUUID().toString();
+ final File file = getFile(ownerId, assetId, assetName);
+ final Asset asset = new MockAsset(assetId, ownerId, assetName, file, null);
+ assets.put(assetId, asset);
+ return asset;
+ }
+
+ @Override
+ public Optional<Asset> deleteAsset(final String id) {
+ final Asset removed = assets.remove(id);
+ if (removed != null && removed.getFile().exists()) {
+ try {
+ Files.delete(removed.getFile().toPath());
+ } catch (final IOException e) {
+ throw new RuntimeException("Failed to delete asset " + id + " from storage file " + removed.getFile().getAbsolutePath(), e);
Review Comment:
```suggestion
throw new UncheckedIOException("Failed to delete asset " + id + " from storage file " + removed.getFile().getAbsolutePath(), e);
```
##########
nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.components.DescribedValue;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.connector.AssetReference;
+import org.apache.nifi.components.connector.ConnectorValueReference;
+import org.apache.nifi.components.connector.FlowUpdateException;
+import org.apache.nifi.components.connector.SecretReference;
+import org.apache.nifi.components.connector.StepConfiguration;
+import org.apache.nifi.flow.VersionedExternalFlow;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+public interface ConnectorTestRunner extends Closeable {
+ String SECRET_PROVIDER_ID = "TestRunnerSecretsManager";
+ String SECRET_PROVIDER_NAME = "TestRunnerSecretsManager";
Review Comment:
Do these constants belong on this interface?
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorAssetManager.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.asset.Asset;
+import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.asset.AssetManagerInitializationContext;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A Mock implementation of AssetManager for use in ConnectorTestRunner tests.
+ * This implementation stores assets in a temporary directory structure.
+ */
+public class MockConnectorAssetManager implements AssetManager {
+
+ private static final String ASSET_STORAGE_LOCATION_PROPERTY = "directory";
+ private static final String DEFAULT_ASSET_STORAGE_LOCATION = "target/mock_connector_assets";
+
+ private final Map<String, Asset> assets = new ConcurrentHashMap<>();
+ private volatile File assetStorageLocation;
+
+ @Override
+ public void initialize(final AssetManagerInitializationContext context) {
+ final String storageLocation = context.getProperties().getOrDefault(ASSET_STORAGE_LOCATION_PROPERTY, DEFAULT_ASSET_STORAGE_LOCATION);
+ assetStorageLocation = new File(storageLocation);
+
+ if (!assetStorageLocation.exists()) {
+ try {
+ Files.createDirectories(assetStorageLocation.toPath());
+ } catch (final IOException e) {
+ throw new RuntimeException("Failed to create asset storage directory: " + storageLocation, e);
Review Comment:
```suggestion
throw new UncheckedIOException("Failed to create asset storage directory: " + storageLocation, e);
```
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockExtensionDiscoveringManager.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.bundle.Bundle;
+import org.apache.nifi.bundle.BundleCoordinate;
+import org.apache.nifi.bundle.BundleDetails;
+import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.nar.InstanceClassLoader;
+import org.apache.nifi.nar.StandardExtensionDiscoveringManager;
+import org.apache.nifi.processor.Processor;
+
+import java.io.File;
+import java.net.URL;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class MockExtensionDiscoveringManager extends StandardExtensionDiscoveringManager {
+ private final ConcurrentMap<String, ClassLoader> mockComponentClassLoaders = new ConcurrentHashMap<>();
+
+ public synchronized void addProcessor(final Class<? extends Processor> mockProcessorClass) {
+ final BundleDetails bundleDetails = new BundleDetails.Builder()
+ .workingDir(new File("target/work/extensions/mock-bundle"))
Review Comment:
Referencing the `target` directory is problematic from the perspective of a clean build environment over multiple runs. Using a JUnit temporary directory provides more independence from the build structure. It also seems like this class should have a configured base work directory, instead of repeating this directory.
##########
nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/resources/flows/Cron_Schedule_Connector.json:
##########
@@ -0,0 +1,410 @@
+{
+ "flowContents": {
+ "identifier": "1800c04e-f9b9-3293-bfc7-b35f43e0706c",
+ "instanceIdentifier": "4f8ca484-019c-1000-955f-68c5defeb22b",
+ "name": "Cron_Schedule_Connector",
+ "comments": "",
+ "position": {
+ "x": -1254.0,
+ "y": -437.70139741897583
+ },
+ "processGroups": [],
+ "remoteProcessGroups": [],
+ "processors": [
+ {
+ "identifier": "4d4160b1-736c-3be4-bc3a-84fc6eb4ad2a",
+ "instanceIdentifier": "499f9781-71c5-367f-aa17-5441bff29de9",
+ "name": "UpdateAttribute",
+ "comments": "",
+ "position": {
+ "x": -360.0,
+ "y": 24.0
+ },
+ "type": "org.apache.nifi.processors.attributes.UpdateAttribute",
+ "bundle": {
+ "group": "org.apache.nifi",
+ "artifact": "nifi-update-attribute-nar",
+ "version": "2026.1.20.21-SNAPSHOT"
Review Comment:
As part of a consistency update, I recommend setting a version of either `2.9.0` or `2.9.0-SNAPSHOT` for all components in Flow JSON files.
##########
nifi-connector-mock-bundle/nifi-connector-mock/src/main/resources/nifi.properties:
##########
@@ -0,0 +1,165 @@
+# 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.
+
+# Core Properties #
+nifi.flow.configuration.file=target/nifi-storage/conf/flow.json.gz
+nifi.flow.configuration.archive.enabled=true
+nifi.flow.configuration.archive.dir=target/nifi-storage/conf/archive/
+nifi.flow.configuration.archive.max.time=30 days
+nifi.flow.configuration.archive.max.storage=500 MB
+nifi.flow.configuration.archive.max.count=
+nifi.flowcontroller.autoResumeState=true
+nifi.flowcontroller.graceful.shutdown.period=10 sec
+nifi.flowservice.writedelay.interval=500 ms
+nifi.administrative.yield.duration=3 sec
+# If a component has no work to do (is "bored"), how long should we wait before checking again for work?
+nifi.bored.yield.duration=10 millis
+nifi.queue.backpressure.count=10000
+nifi.queue.backpressure.size=1 GB
+
+nifi.nar.library.directory=./lib
+nifi.nar.library.autoload.directory=./extensions
+nifi.nar.working.directory=./work/nar/
+
+#####################
+# Python Extensions #
+#####################
+# Uncomment in order to enable Python Extensions.
+#nifi.python.command=python
+nifi.python.framework.source.directory=./python/framework
+nifi.python.extensions.source.directory.default=./python/extensions
+nifi.python.working.directory=./work/python
+nifi.python.max.processes=100
+nifi.python.max.processes.per.extension.type=10
+
+####################
+# State Management #
+####################
+nifi.state.management.configuration.file=target/nifi-storage/state-management.xml
+# The ID of the local state provider
+nifi.state.management.provider.local=local-provider
+
+# Database Settings
+nifi.database.directory=target/nifi-storage/database_repository
+
+# FlowFile Repository
+nifi.flowfile.repository.implementation=org.apache.nifi.controller.repository.WriteAheadFlowFileRepository
+nifi.flowfile.repository.wal.implementation=org.apache.nifi.wali.SequentialAccessWriteAheadLog
+nifi.flowfile.repository.directory=target/nifi-storage/flowfile_repository
+nifi.flowfile.repository.checkpoint.interval=1 mins
+nifi.flowfile.repository.always.sync=false
+
+nifi.swap.manager.implementation=org.apache.nifi.controller.FileSystemSwapManager
+nifi.queue.swap.threshold=20000
+nifi.swap.in.period=5 sec
+nifi.swap.in.threads=1
+nifi.swap.out.period=5 sec
+nifi.swap.out.threads=4
+
+# Content Repository
+nifi.content.repository.implementation=org.apache.nifi.controller.repository.FileSystemRepository
+nifi.content.claim.max.appendable.size=50 KB
+nifi.content.repository.directory.default=target/nifi-storage/content_repository
+nifi.content.repository.archive.max.retention.period=12 hours
+nifi.content.repository.archive.max.usage.percentage=90%
+nifi.content.repository.archive.enabled=false
+nifi.content.repository.always.sync=false
+
+# Provenance Repository Properties
+nifi.provenance.repository.implementation=org.apache.nifi.provenance.VolatileProvenanceRepository
+
+# Volatile Provenance Respository Properties
+nifi.provenance.repository.buffer.size=1000
+
+# Component Status Repository
+nifi.components.status.repository.implementation=org.apache.nifi.controller.status.history.VolatileComponentStatusRepository
+nifi.components.status.repository.buffer.size=1440
+nifi.components.status.snapshot.frequency=1 min
+
+# NAR Persistence Provider Properties
+nifi.nar.persistence.provider.properties.directory=target/nifi-storage/nar_repository
+
+# Asset Management
+nifi.asset.manager.properties.directory=target/nifi-storage/assets
+
+# Connector Asset Manager
+nifi.connector.asset.manager.implementation=org.apache.nifi.mock.connector.server.MockConnectorAssetManager
+
+# Secrets Manager
+nifi.secrets.manager.implementation=org.apache.nifi.mock.connector.server.secrets.ConnectorTestRunnerSecretsManager
+
+# Site to Site properties
+nifi.remote.input.host=localhost
+nifi.remote.input.secure=true
+nifi.remote.input.socket.port=7780
+nifi.remote.input.http.enabled=true
+nifi.remote.input.http.transaction.ttl=30 sec
+nifi.remote.contents.cache.expiration=30 secs
+
+# web properties #
+nifi.web.war.directory=./lib
+nifi.web.http.host=
+nifi.web.http.port=
+nifi.web.http.network.interface.default=
+nifi.web.https.host=localhost
+nifi.web.https.port=0
+nifi.web.https.network.interface.default=
+nifi.web.jetty.working.directory=target/work/jetty
+nifi.web.jetty.threads=200
+nifi.web.max.header.size=16 KB
+nifi.web.proxy.context.path=
+nifi.web.proxy.host=
+
+# security properties #
+nifi.sensitive.props.key=nifi-connector-mock-tests
+nifi.sensitive.props.algorithm=NIFI_PBKDF2_AES_GCM_256
+
+nifi.security.keystore=
+nifi.security.keystoreType=PKCS12
+nifi.security.keystorePasswd=
+nifi.security.keyPasswd=
+nifi.security.truststore=
+nifi.security.truststoreType=PKCS12
+nifi.security.truststorePasswd=
+nifi.security.user.authorizer=system-test-authorizer
+nifi.security.user.login.identity.provider=
+
+# cluster common properties (all nodes must have same values) #
+nifi.cluster.protocol.heartbeat.interval=5 sec
+nifi.cluster.protocol.is.secure=true
+
+# cluster node properties (only configure for cluster nodes) #
+nifi.cluster.is.node=false
+nifi.cluster.node.address=
+nifi.cluster.node.protocol.port=
+nifi.cluster.node.protocol.threads=10
+nifi.cluster.node.protocol.max.threads=50
+nifi.cluster.node.event.history.size=25
+nifi.cluster.node.connection.timeout=5 sec
+nifi.cluster.node.read.timeout=5 sec
+nifi.cluster.node.max.concurrent.requests=100
+nifi.cluster.firewall.file=
+nifi.cluster.flow.election.max.wait.time=5 mins
+nifi.cluster.flow.election.max.candidates=
+
+# cluster load balancing properties #
+nifi.cluster.load.balance.host=
+nifi.cluster.load.balance.port=6342
+nifi.cluster.load.balance.connections.per.node=4
+nifi.cluster.load.balance.max.thread.count=8
+nifi.cluster.load.balance.comms.timeout=30 sec
Review Comment:
Highlighting these cluster properties, but there appear to be many unnecessary properties in this file that could be removed so that defaults are used.
##########
nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/pom.xml:
##########
@@ -0,0 +1,238 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-kafka-to-s3-bundle</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-kafka-to-s3-integration-tests</artifactId>
+
+ <repositories>
+ <repository>
+ <id>confluent</id>
+ <url>https://packages.confluent.io/maven/</url>
+ </repository>
+ </repositories>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock-api</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+
+ <!-- Test Containers for Kafka and S3 -->
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers-kafka</artifactId>
+ <version>2.0.1</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>localstack</artifactId>
+ <version>1.20.4</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Kafka client for producing test data -->
+ <!-- Override vulnerable kafka clients -->
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>4.1.1</version>
+ <exclusions>
+ <!-- Excluded and replaced with at.yawk.lz4 version -->
+ <exclusion>
+ <groupId>org.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>at.yawk.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ <version>1.10.1</version>
+ </dependency>
+
+ <!-- Confluent Schema Registry and Avro dependencies for testing -->
+ <dependency>
+ <groupId>io.confluent</groupId>
+ <artifactId>kafka-avro-serializer</artifactId>
+ <version>7.8.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.confluent</groupId>
+ <artifactId>kafka-schema-registry-client</artifactId>
+ <version>7.8.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <version>1.11.3</version>
Review Comment:
The Avro version should come from the parent Maven configuration
##########
nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-connector/src/main/java/org/apache/nifi/connectors/kafkas3/S3Step.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.connectors.kafkas3;
+
+import org.apache.nifi.components.connector.ConfigurationStep;
+import org.apache.nifi.components.connector.ConnectorPropertyDescriptor;
+import org.apache.nifi.components.connector.ConnectorPropertyGroup;
+import org.apache.nifi.components.connector.PropertyType;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.List;
+
+public class S3Step {
+ public static final String S3_STEP_NAME = "S3 Configuration";
+ public static final String ACCESS_KEY_ID_SECRET_KEY = "Access Key ID and Secret Key";
+ public static final String DEFAULT_CREDENTIALS = "Default AWS Credentials";
+
+
+ public static final ConnectorPropertyDescriptor S3_BUCKET = new ConnectorPropertyDescriptor.Builder()
+ .name("S3 Bucket")
+ .description("The name of the S3 bucket to write data to.")
+ .required(true)
+ .build();
+
+ public static final ConnectorPropertyDescriptor S3_REGION = new ConnectorPropertyDescriptor.Builder()
+ .name("S3 Region")
+ .description("The AWS region where the S3 bucket is located.")
+ .allowableValuesFetchable(true)
+ .required(true)
+ .build();
+
+ public static final ConnectorPropertyDescriptor S3_DATA_FORMAT = new ConnectorPropertyDescriptor.Builder()
+ .name("S3 Data Format")
+ .description("The format to use when writing data to S3.")
+ .required(true)
+ .defaultValue("JSON")
+ .allowableValues("Avro", "JSON")
+ .build();
+
+ public static final ConnectorPropertyDescriptor S3_PREFIX = new ConnectorPropertyDescriptor.Builder()
+ .name("S3 Prefix")
+ .description("An optional prefix to prepend to all object keys written to the S3 bucket.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final ConnectorPropertyDescriptor S3_ENDPOINT_OVERRIDE_URL = new ConnectorPropertyDescriptor.Builder()
+ .name("S3 Endpoint Override URL")
+ .description("An optional endpoint URL to use instead of the default AWS S3 endpoint. " +
+ "This can be used to connect to S3-compatible storage systems but should be left unset for connecting to S3.")
+ .required(false)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ public static final ConnectorPropertyDescriptor TARGET_OBJECT_SIZE = new ConnectorPropertyDescriptor.Builder()
+ .name("Target Object Size")
+ .description("The target size for each object written to S3. The connector will attempt to " +
+ "combine messages until this size is reached before writing an object to S3. Note that this size is approximate " +
+ "and may vary from object to object.")
Review Comment:
Descriptions can use multi-line strings.
##########
nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/pom.xml:
##########
@@ -0,0 +1,238 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-kafka-to-s3-bundle</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-kafka-to-s3-integration-tests</artifactId>
+
+ <repositories>
+ <repository>
+ <id>confluent</id>
+ <url>https://packages.confluent.io/maven/</url>
+ </repository>
+ </repositories>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock-api</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+
+ <!-- Test Containers for Kafka and S3 -->
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers-kafka</artifactId>
+ <version>2.0.1</version>
Review Comment:
The testcontainers version should come from the root Maven configuration.
##########
nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/pom.xml:
##########
@@ -0,0 +1,238 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-kafka-to-s3-bundle</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-kafka-to-s3-integration-tests</artifactId>
+
+ <repositories>
+ <repository>
+ <id>confluent</id>
+ <url>https://packages.confluent.io/maven/</url>
+ </repository>
+ </repositories>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock-api</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+
+ <!-- Test Containers for Kafka and S3 -->
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers-kafka</artifactId>
+ <version>2.0.1</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>localstack</artifactId>
+ <version>1.20.4</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Kafka client for producing test data -->
+ <!-- Override vulnerable kafka clients -->
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>4.1.1</version>
+ <exclusions>
+ <!-- Excluded and replaced with at.yawk.lz4 version -->
+ <exclusion>
+ <groupId>org.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>at.yawk.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ <version>1.10.1</version>
+ </dependency>
+
+ <!-- Confluent Schema Registry and Avro dependencies for testing -->
+ <dependency>
+ <groupId>io.confluent</groupId>
+ <artifactId>kafka-avro-serializer</artifactId>
+ <version>7.8.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.confluent</groupId>
+ <artifactId>kafka-schema-registry-client</artifactId>
+ <version>7.8.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <version>1.11.3</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- AWS SDK for S3 verification in tests -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>s3</artifactId>
+ <version>2.36.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>url-connection-client</artifactId>
+ <version>2.36.0</version>
Review Comment:
The AWS SDK client version should come from the parent Maven configuration
##########
nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/pom.xml:
##########
@@ -0,0 +1,238 @@
+<?xml version="1.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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-kafka-to-s3-bundle</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-kafka-to-s3-integration-tests</artifactId>
+
+ <repositories>
+ <repository>
+ <id>confluent</id>
+ <url>https://packages.confluent.io/maven/</url>
+ </repository>
+ </repositories>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock-api</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-connector-mock</artifactId>
+ <version>2.9.0-SNAPSHOT</version>
+ </dependency>
+
+ <!-- Test Containers for Kafka and S3 -->
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers-kafka</artifactId>
+ <version>2.0.1</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>localstack</artifactId>
+ <version>1.20.4</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Kafka client for producing test data -->
+ <!-- Override vulnerable kafka clients -->
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>4.1.1</version>
+ <exclusions>
+ <!-- Excluded and replaced with at.yawk.lz4 version -->
+ <exclusion>
+ <groupId>org.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>at.yawk.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ <version>1.10.1</version>
+ </dependency>
+
+ <!-- Confluent Schema Registry and Avro dependencies for testing -->
+ <dependency>
+ <groupId>io.confluent</groupId>
+ <artifactId>kafka-avro-serializer</artifactId>
+ <version>7.8.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.confluent</groupId>
+ <artifactId>kafka-schema-registry-client</artifactId>
+ <version>7.8.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <version>1.11.3</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- AWS SDK for S3 verification in tests -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>s3</artifactId>
+ <version>2.36.0</version>
Review Comment:
The S3 version should come from the parent Maven configuration
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2914802837
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorAssetManager.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.asset.Asset;
+import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.asset.AssetManagerInitializationContext;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A Mock implementation of AssetManager for use in ConnectorTestRunner tests.
+ * This implementation stores assets in a temporary directory structure.
+ */
+public class MockConnectorAssetManager implements AssetManager {
+
+ private static final String ASSET_STORAGE_LOCATION_PROPERTY = "directory";
+ private static final String DEFAULT_ASSET_STORAGE_LOCATION = "target/mock_connector_assets";
+
+ private final Map<String, Asset> assets = new ConcurrentHashMap<>();
+ private volatile File assetStorageLocation;
+
+ @Override
+ public void initialize(final AssetManagerInitializationContext context) {
+ final String storageLocation = context.getProperties().getOrDefault(ASSET_STORAGE_LOCATION_PROPERTY, DEFAULT_ASSET_STORAGE_LOCATION);
+ assetStorageLocation = new File(storageLocation);
+
+ if (!assetStorageLocation.exists()) {
+ try {
+ Files.createDirectories(assetStorageLocation.toPath());
+ } catch (final IOException e) {
+ throw new RuntimeException("Failed to create asset storage directory: " + storageLocation, e);
+ }
+ }
+ }
+
+ @Override
+ public Asset createAsset(final String ownerId, final String assetName, final InputStream contents) throws IOException {
+ final String assetId = UUID.randomUUID().toString();
+ return saveAsset(ownerId, assetId, assetName, contents);
+ }
+
+ @Override
+ public Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents) throws IOException {
+ final File assetFile = getFile(ownerId, assetId, assetName);
+ final File parentDir = assetFile.getParentFile();
+
+ if (!parentDir.exists()) {
+ Files.createDirectories(parentDir.toPath());
+ }
+
+ Files.copy(contents, assetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/89)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/asset/StandardAssetManager.java:
##########
@@ -155,14 +139,43 @@
return Optional.of(removed);
}
- private String createAssetId(final String parameterContextId, final String assetName) {
- final String seed = parameterContextId + "/" + assetName;
+ private Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents, final File assetFile) throws IOException {
+ final File dir = assetFile.getParentFile();
+ if (!dir.exists()) {
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/73)
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorAssetManager.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.asset.Asset;
+import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.asset.AssetManagerInitializationContext;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A Mock implementation of AssetManager for use in ConnectorTestRunner tests.
+ * This implementation stores assets in a temporary directory structure.
+ */
+public class MockConnectorAssetManager implements AssetManager {
+
+ private static final String ASSET_STORAGE_LOCATION_PROPERTY = "directory";
+ private static final String DEFAULT_ASSET_STORAGE_LOCATION = "target/mock_connector_assets";
+
+ private final Map<String, Asset> assets = new ConcurrentHashMap<>();
+ private volatile File assetStorageLocation;
+
+ @Override
+ public void initialize(final AssetManagerInitializationContext context) {
+ final String storageLocation = context.getProperties().getOrDefault(ASSET_STORAGE_LOCATION_PROPERTY, DEFAULT_ASSET_STORAGE_LOCATION);
+ assetStorageLocation = new File(storageLocation);
+
+ if (!assetStorageLocation.exists()) {
+ try {
+ Files.createDirectories(assetStorageLocation.toPath());
+ } catch (final IOException e) {
+ throw new RuntimeException("Failed to create asset storage directory: " + storageLocation, e);
+ }
+ }
+ }
+
+ @Override
+ public Asset createAsset(final String ownerId, final String assetName, final InputStream contents) throws IOException {
+ final String assetId = UUID.randomUUID().toString();
+ return saveAsset(ownerId, assetId, assetName, contents);
+ }
+
+ @Override
+ public Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents) throws IOException {
+ final File assetFile = getFile(ownerId, assetId, assetName);
+ final File parentDir = assetFile.getParentFile();
+
+ if (!parentDir.exists()) {
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/87)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/asset/StandardAssetManager.java:
##########
@@ -155,14 +139,43 @@
return Optional.of(removed);
}
- private String createAssetId(final String parameterContextId, final String assetName) {
- final String seed = parameterContextId + "/" + assetName;
+ private Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents, final File assetFile) throws IOException {
+ final File dir = assetFile.getParentFile();
+ if (!dir.exists()) {
+ try {
+ Files.createDirectories(dir.toPath());
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/74)
##########
nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorAssetManager.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.asset.Asset;
+import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.asset.AssetManagerInitializationContext;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A Mock implementation of AssetManager for use in ConnectorTestRunner tests.
+ * This implementation stores assets in a temporary directory structure.
+ */
+public class MockConnectorAssetManager implements AssetManager {
+
+ private static final String ASSET_STORAGE_LOCATION_PROPERTY = "directory";
+ private static final String DEFAULT_ASSET_STORAGE_LOCATION = "target/mock_connector_assets";
+
+ private final Map<String, Asset> assets = new ConcurrentHashMap<>();
+ private volatile File assetStorageLocation;
+
+ @Override
+ public void initialize(final AssetManagerInitializationContext context) {
+ final String storageLocation = context.getProperties().getOrDefault(ASSET_STORAGE_LOCATION_PROPERTY, DEFAULT_ASSET_STORAGE_LOCATION);
+ assetStorageLocation = new File(storageLocation);
+
+ if (!assetStorageLocation.exists()) {
+ try {
+ Files.createDirectories(assetStorageLocation.toPath());
+ } catch (final IOException e) {
+ throw new RuntimeException("Failed to create asset storage directory: " + storageLocation, e);
+ }
+ }
+ }
+
+ @Override
+ public Asset createAsset(final String ownerId, final String assetName, final InputStream contents) throws IOException {
+ final String assetId = UUID.randomUUID().toString();
+ return saveAsset(ownerId, assetId, assetName, contents);
+ }
+
+ @Override
+ public Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents) throws IOException {
+ final File assetFile = getFile(ownerId, assetId, assetName);
+ final File parentDir = assetFile.getParentFile();
+
+ if (!parentDir.exists()) {
+ Files.createDirectories(parentDir.toPath());
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/88)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/asset/StandardAssetManager.java:
##########
@@ -155,14 +139,43 @@
return Optional.of(removed);
}
- private String createAssetId(final String parameterContextId, final String assetName) {
- final String seed = parameterContextId + "/" + assetName;
+ private Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents, final File assetFile) throws IOException {
+ final File dir = assetFile.getParentFile();
+ if (!dir.exists()) {
+ try {
+ Files.createDirectories(dir.toPath());
+ } catch (final IOException ioe) {
+ throw new IOException("Could not create directory in order to store asset", ioe);
+ }
+ }
+
+ // Write contents to a temporary file, then move it to the final location.
+ // This allows us to avoid a situation where we upload a file, then we attempt to overwrite it but fail, leaving a corrupt asset.
+ final File tempFile = new File(dir, assetFile.getName() + ".tmp");
+ logger.debug("Writing temp asset file [{}]", tempFile.getAbsolutePath());
+
+ try {
+ Files.copy(contents, tempFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/75)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/asset/StandardConnectorAssetManager.java:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.asset;
+
+import org.apache.nifi.nar.FileDigestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * AssetManager implementation for Connectors. This differs from the StandardAssetManager used for Parameter Contexts in that
+ * Connectors want every asset added to be treated as a new asset, even if the same filename is being saved to the same Connector.
+ * This is because all changes to the Connector needs to be stored separately in the working configuration until applied and can
+ * not interfere with the active configuration.
+ */
+public class StandardConnectorAssetManager implements AssetManager {
+
+ private static final Logger logger = LoggerFactory.getLogger(StandardConnectorAssetManager.class);
+
+ public static final String ASSET_STORAGE_LOCATION_PROPERTY = "directory";
+ public static final String DEFAULT_ASSET_STORAGE_LOCATION = "./connector_assets";
+
+ private volatile File assetStorageLocation;
+ private final Map<String, Asset> assets = new ConcurrentHashMap<>();
+
+ @Override
+ public void initialize(final AssetManagerInitializationContext context) {
+ final String storageLocation = getStorageLocation(context);
+
+ assetStorageLocation = new File(storageLocation);
+ if (!assetStorageLocation.exists()) {
+ try {
+ Files.createDirectories(assetStorageLocation.toPath());
+ } catch (IOException e) {
+ throw new RuntimeException("The Connector Asset Manager's [%s] property is set to [%s] but the directory does not exist and cannot be created"
+ .formatted(ASSET_STORAGE_LOCATION_PROPERTY, storageLocation), e);
+ }
+ }
+
+ try {
+ recoverLocalAssets();
+ } catch (final IOException e) {
+ throw new RuntimeException("Unable to access connector assets", e);
+ }
+ }
+
+ @Override
+ public Asset createAsset(final String connectorId, final String assetName, final InputStream contents) throws IOException {
+ final String assetId = generateAssetId();
+ final File assetFile = getFile(connectorId, assetId, assetName);
+ return saveAsset(connectorId, assetId, assetName, contents, assetFile);
+ }
+
+ @Override
+ public Asset saveAsset(final String connectorId, final String assetId, final String assetName, final InputStream contents) throws IOException {
+ final File assetFile = getFile(connectorId, assetId, assetName);
+ return saveAsset(connectorId, assetId, assetName, contents, assetFile);
+ }
+
+ @Override
+ public Optional<Asset> getAsset(final String id) {
+ return Optional.ofNullable(assets.get(id));
+ }
+
+ @Override
+ public List<Asset> getAssets(final String connectorId) {
+ final List<Asset> allAssets = new ArrayList<>(assets.values());
+ final List<Asset> ownerAssets = new ArrayList<>();
+ for (final Asset asset : allAssets) {
+ if (asset.getOwnerIdentifier().equals(connectorId)) {
+ ownerAssets.add(asset);
+ }
+ }
+ return ownerAssets;
+ }
+
+ @Override
+ public Asset createMissingAsset(final String connectorId, final String assetName) {
+ final String assetId = generateAssetId();
+ final File file = getFile(connectorId, assetId, assetName);
+ final Asset asset = new StandardAsset(assetId, connectorId, assetName, file, null);
+ assets.put(assetId, asset);
+ return asset;
+ }
+
+ @Override
+ public Optional<Asset> deleteAsset(final String id) {
+ final Asset removed = assets.remove(id);
+ if (removed == null) {
+ return Optional.empty();
+ }
+
+ final File assetFile = removed.getFile();
+ if (assetFile.exists()) {
+ deleteFile(assetFile);
+
+ final File assetIdDir = assetFile.getParentFile();
+ deleteFile(assetIdDir);
+
+ final File connectorDir = assetIdDir.getParentFile();
+ final File[] children = connectorDir.listFiles();
+ if (children != null && children.length == 0) {
+ deleteFile(connectorDir);
+ }
+ }
+
+ return Optional.of(removed);
+ }
+
+ private Asset saveAsset(final String connectorId, final String assetId, final String assetName, final InputStream contents, final File assetFile) throws IOException {
+ final File dir = assetFile.getParentFile();
+ if (!dir.exists()) {
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/90)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/asset/StandardAssetManager.java:
##########
@@ -155,14 +139,43 @@
return Optional.of(removed);
}
- private String createAssetId(final String parameterContextId, final String assetName) {
- final String seed = parameterContextId + "/" + assetName;
+ private Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents, final File assetFile) throws IOException {
+ final File dir = assetFile.getParentFile();
+ if (!dir.exists()) {
+ try {
+ Files.createDirectories(dir.toPath());
+ } catch (final IOException ioe) {
+ throw new IOException("Could not create directory in order to store asset", ioe);
+ }
+ }
+
+ // Write contents to a temporary file, then move it to the final location.
+ // This allows us to avoid a situation where we upload a file, then we attempt to overwrite it but fail, leaving a corrupt asset.
+ final File tempFile = new File(dir, assetFile.getName() + ".tmp");
+ logger.debug("Writing temp asset file [{}]", tempFile.getAbsolutePath());
+
+ try {
+ Files.copy(contents, tempFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
+ } catch (final Exception e) {
+ throw new IOException("Failed to write asset to file " + tempFile.getAbsolutePath(), e);
+ }
+
+ Files.move(tempFile.toPath(), assetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/94)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/asset/StandardConnectorAssetManager.java:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.asset;
+
+import org.apache.nifi.nar.FileDigestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * AssetManager implementation for Connectors. This differs from the StandardAssetManager used for Parameter Contexts in that
+ * Connectors want every asset added to be treated as a new asset, even if the same filename is being saved to the same Connector.
+ * This is because all changes to the Connector needs to be stored separately in the working configuration until applied and can
+ * not interfere with the active configuration.
+ */
+public class StandardConnectorAssetManager implements AssetManager {
+
+ private static final Logger logger = LoggerFactory.getLogger(StandardConnectorAssetManager.class);
+
+ public static final String ASSET_STORAGE_LOCATION_PROPERTY = "directory";
+ public static final String DEFAULT_ASSET_STORAGE_LOCATION = "./connector_assets";
+
+ private volatile File assetStorageLocation;
+ private final Map<String, Asset> assets = new ConcurrentHashMap<>();
+
+ @Override
+ public void initialize(final AssetManagerInitializationContext context) {
+ final String storageLocation = getStorageLocation(context);
+
+ assetStorageLocation = new File(storageLocation);
+ if (!assetStorageLocation.exists()) {
+ try {
+ Files.createDirectories(assetStorageLocation.toPath());
+ } catch (IOException e) {
+ throw new RuntimeException("The Connector Asset Manager's [%s] property is set to [%s] but the directory does not exist and cannot be created"
+ .formatted(ASSET_STORAGE_LOCATION_PROPERTY, storageLocation), e);
+ }
+ }
+
+ try {
+ recoverLocalAssets();
+ } catch (final IOException e) {
+ throw new RuntimeException("Unable to access connector assets", e);
+ }
+ }
+
+ @Override
+ public Asset createAsset(final String connectorId, final String assetName, final InputStream contents) throws IOException {
+ final String assetId = generateAssetId();
+ final File assetFile = getFile(connectorId, assetId, assetName);
+ return saveAsset(connectorId, assetId, assetName, contents, assetFile);
+ }
+
+ @Override
+ public Asset saveAsset(final String connectorId, final String assetId, final String assetName, final InputStream contents) throws IOException {
+ final File assetFile = getFile(connectorId, assetId, assetName);
+ return saveAsset(connectorId, assetId, assetName, contents, assetFile);
+ }
+
+ @Override
+ public Optional<Asset> getAsset(final String id) {
+ return Optional.ofNullable(assets.get(id));
+ }
+
+ @Override
+ public List<Asset> getAssets(final String connectorId) {
+ final List<Asset> allAssets = new ArrayList<>(assets.values());
+ final List<Asset> ownerAssets = new ArrayList<>();
+ for (final Asset asset : allAssets) {
+ if (asset.getOwnerIdentifier().equals(connectorId)) {
+ ownerAssets.add(asset);
+ }
+ }
+ return ownerAssets;
+ }
+
+ @Override
+ public Asset createMissingAsset(final String connectorId, final String assetName) {
+ final String assetId = generateAssetId();
+ final File file = getFile(connectorId, assetId, assetName);
+ final Asset asset = new StandardAsset(assetId, connectorId, assetName, file, null);
+ assets.put(assetId, asset);
+ return asset;
+ }
+
+ @Override
+ public Optional<Asset> deleteAsset(final String id) {
+ final Asset removed = assets.remove(id);
+ if (removed == null) {
+ return Optional.empty();
+ }
+
+ final File assetFile = removed.getFile();
+ if (assetFile.exists()) {
+ deleteFile(assetFile);
+
+ final File assetIdDir = assetFile.getParentFile();
+ deleteFile(assetIdDir);
+
+ final File connectorDir = assetIdDir.getParentFile();
+ final File[] children = connectorDir.listFiles();
+ if (children != null && children.length == 0) {
+ deleteFile(connectorDir);
+ }
+ }
+
+ return Optional.of(removed);
+ }
+
+ private Asset saveAsset(final String connectorId, final String assetId, final String assetName, final InputStream contents, final File assetFile) throws IOException {
+ final File dir = assetFile.getParentFile();
+ if (!dir.exists()) {
+ try {
+ Files.createDirectories(dir.toPath());
+ } catch (final IOException ioe) {
+ throw new IOException("Could not create directory in order to store connector asset", ioe);
+ }
+ }
+
+ try {
+ Files.copy(contents, assetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/92)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/asset/StandardConnectorAssetManager.java:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.asset;
+
+import org.apache.nifi.nar.FileDigestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * AssetManager implementation for Connectors. This differs from the StandardAssetManager used for Parameter Contexts in that
+ * Connectors want every asset added to be treated as a new asset, even if the same filename is being saved to the same Connector.
+ * This is because all changes to the Connector needs to be stored separately in the working configuration until applied and can
+ * not interfere with the active configuration.
+ */
+public class StandardConnectorAssetManager implements AssetManager {
+
+ private static final Logger logger = LoggerFactory.getLogger(StandardConnectorAssetManager.class);
+
+ public static final String ASSET_STORAGE_LOCATION_PROPERTY = "directory";
+ public static final String DEFAULT_ASSET_STORAGE_LOCATION = "./connector_assets";
+
+ private volatile File assetStorageLocation;
+ private final Map<String, Asset> assets = new ConcurrentHashMap<>();
+
+ @Override
+ public void initialize(final AssetManagerInitializationContext context) {
+ final String storageLocation = getStorageLocation(context);
+
+ assetStorageLocation = new File(storageLocation);
+ if (!assetStorageLocation.exists()) {
+ try {
+ Files.createDirectories(assetStorageLocation.toPath());
+ } catch (IOException e) {
+ throw new RuntimeException("The Connector Asset Manager's [%s] property is set to [%s] but the directory does not exist and cannot be created"
+ .formatted(ASSET_STORAGE_LOCATION_PROPERTY, storageLocation), e);
+ }
+ }
+
+ try {
+ recoverLocalAssets();
+ } catch (final IOException e) {
+ throw new RuntimeException("Unable to access connector assets", e);
+ }
+ }
+
+ @Override
+ public Asset createAsset(final String connectorId, final String assetName, final InputStream contents) throws IOException {
+ final String assetId = generateAssetId();
+ final File assetFile = getFile(connectorId, assetId, assetName);
+ return saveAsset(connectorId, assetId, assetName, contents, assetFile);
+ }
+
+ @Override
+ public Asset saveAsset(final String connectorId, final String assetId, final String assetName, final InputStream contents) throws IOException {
+ final File assetFile = getFile(connectorId, assetId, assetName);
+ return saveAsset(connectorId, assetId, assetName, contents, assetFile);
+ }
+
+ @Override
+ public Optional<Asset> getAsset(final String id) {
+ return Optional.ofNullable(assets.get(id));
+ }
+
+ @Override
+ public List<Asset> getAssets(final String connectorId) {
+ final List<Asset> allAssets = new ArrayList<>(assets.values());
+ final List<Asset> ownerAssets = new ArrayList<>();
+ for (final Asset asset : allAssets) {
+ if (asset.getOwnerIdentifier().equals(connectorId)) {
+ ownerAssets.add(asset);
+ }
+ }
+ return ownerAssets;
+ }
+
+ @Override
+ public Asset createMissingAsset(final String connectorId, final String assetName) {
+ final String assetId = generateAssetId();
+ final File file = getFile(connectorId, assetId, assetName);
+ final Asset asset = new StandardAsset(assetId, connectorId, assetName, file, null);
+ assets.put(assetId, asset);
+ return asset;
+ }
+
+ @Override
+ public Optional<Asset> deleteAsset(final String id) {
+ final Asset removed = assets.remove(id);
+ if (removed == null) {
+ return Optional.empty();
+ }
+
+ final File assetFile = removed.getFile();
+ if (assetFile.exists()) {
+ deleteFile(assetFile);
+
+ final File assetIdDir = assetFile.getParentFile();
+ deleteFile(assetIdDir);
+
+ final File connectorDir = assetIdDir.getParentFile();
+ final File[] children = connectorDir.listFiles();
+ if (children != null && children.length == 0) {
+ deleteFile(connectorDir);
+ }
+ }
+
+ return Optional.of(removed);
+ }
+
+ private Asset saveAsset(final String connectorId, final String assetId, final String assetName, final InputStream contents, final File assetFile) throws IOException {
+ final File dir = assetFile.getParentFile();
+ if (!dir.exists()) {
+ try {
+ Files.createDirectories(dir.toPath());
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/91)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/asset/StandardAssetManager.java:
##########
@@ -155,14 +139,43 @@
return Optional.of(removed);
}
- private String createAssetId(final String parameterContextId, final String assetName) {
- final String seed = parameterContextId + "/" + assetName;
+ private Asset saveAsset(final String ownerId, final String assetId, final String assetName, final InputStream contents, final File assetFile) throws IOException {
+ final File dir = assetFile.getParentFile();
+ if (!dir.exists()) {
+ try {
+ Files.createDirectories(dir.toPath());
+ } catch (final IOException ioe) {
+ throw new IOException("Could not create directory in order to store asset", ioe);
+ }
+ }
+
+ // Write contents to a temporary file, then move it to the final location.
+ // This allows us to avoid a situation where we upload a file, then we attempt to overwrite it but fail, leaving a corrupt asset.
+ final File tempFile = new File(dir, assetFile.getName() + ".tmp");
+ logger.debug("Writing temp asset file [{}]", tempFile.getAbsolutePath());
+
+ try {
+ Files.copy(contents, tempFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
+ } catch (final Exception e) {
+ throw new IOException("Failed to write asset to file " + tempFile.getAbsolutePath(), e);
+ }
+
+ Files.move(tempFile.toPath(), assetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
[Show more details](https://github.com/apache/nifi/security/code-scanning/93)
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/manifest/StandardRuntimeManifestService.java:
##########
@@ -401,6 +401,51 @@
return additionalDetailsMap;
}
+ @Override
+ public Map<String, File> discoverStepDocumentation(final String group, final String artifact, final String version, final String connectorType) {
+ final BundleCoordinate bundleCoordinate = new BundleCoordinate(group, artifact, version);
+ final Bundle bundle = extensionManager.getBundle(bundleCoordinate);
+
+ if (bundle == null) {
+ throw new ResourceNotFoundException("Unable to find bundle [" + bundleCoordinate + "]");
+ }
+
+ return discoverStepDocumentation(bundle.getBundleDetails(), connectorType);
+ }
+
+ private Map<String, File> discoverStepDocumentation(final BundleDetails bundleDetails, final String connectorType) {
+ final Map<String, File> stepDocsMap = new LinkedHashMap<>();
+
+ final File stepDocsDir = new File(bundleDetails.getWorkingDirectory(), "META-INF/docs/steps/" + connectorType);
+ if (!stepDocsDir.exists()) {
Review Comment:
## Uncontrolled data used in path expression
This path depends on a [user-provided value](1).
This path depends on a [user-provided value](2).
This path depends on a [user-provided value](3).
[Show more details](https://github.com/apache/nifi/security/code-scanning/95)
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2982594630
##########
nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/HashMapSnapshot.java:
##########
@@ -175,14 +175,18 @@ public void update(final Collection<T> records) {
switch (updateType) {
case DELETE:
- recordMap.remove(recordId);
+ if (recordId != null) {
Review Comment:
There are actually 2 changes here, really: The null-checks here in the DELETE/SWAP_OUT/etc. cases; and the SWAP_FILE_DELETED/SWAP_FILE_RENAMED branches added.
The null-checks are actually not needed. I will revert that. The only records that should come in here with null IDs are the SWAP_FILE_DELETE, SWAP-FILE_RENAMED types. Those actually are necessary for this PR. The reason they are introduced here is that this introduces the new capability of removing FlowFiles from the flow selectively, and this could mean removing some but not all FlowFiles from a Swap File OR removing all FlowFiles in a Swap File. This was previously never done (swap files were swapped in entirely, even when a queue was drained).
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2984741828
##########
nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.mock.connector.server;
+
+import org.apache.nifi.components.DescribedValue;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.connector.AssetReference;
+import org.apache.nifi.components.connector.ConnectorValueReference;
+import org.apache.nifi.components.connector.FlowUpdateException;
+import org.apache.nifi.components.connector.SecretReference;
+import org.apache.nifi.components.connector.StepConfiguration;
+import org.apache.nifi.flow.VersionedExternalFlow;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+
+public interface ConnectorTestRunner extends Closeable {
+ String SECRET_PROVIDER_ID = "TestRunnerSecretsManager";
+ String SECRET_PROVIDER_NAME = "TestRunnerSecretsManager";
Review Comment:
Nope. Refactoring faux pas :)
--
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
Re: [PR] NIFI-15258: Initial implementation of Connectors [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #10993:
URL: https://github.com/apache/nifi/pull/10993#discussion_r2991232717
##########
nifi-commons/nifi-write-ahead-log/src/main/java/org/apache/nifi/wali/HashMapSnapshot.java:
##########
@@ -175,14 +175,18 @@ public void update(final Collection<T> records) {
switch (updateType) {
case DELETE:
- recordMap.remove(recordId);
+ if (recordId != null) {
Review Comment:
Thanks for clarifying and pointing to the reason for the change, sounds good.
--
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