You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "tanmaya-panda1 (via GitHub)" <gi...@apache.org> on 2023/04/04 15:37:30 UTC
[GitHub] [nifi] tanmaya-panda1 opened a new pull request, #7122: NIFI-8294 ADX as source for various NIFI destinations
tanmaya-panda1 opened a new pull request, #7122:
URL: https://github.com/apache/nifi/pull/7122
<!-- 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-8294](https://issues.apache.org/jira/browse/NIFI-8294)
# 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 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 `mvn clean install -P contrib-check`
- [ ] JDK 11
- [ ] JDK 17
### Licensing
- [ ] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
- [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [ ] Documentation formatting appears as expected in rendered files
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tanmaya-panda1 commented on pull request #7122: NIFI-8294 Add Query Processor for Azure Data Explorer
Posted by "tanmaya-panda1 (via GitHub)" <gi...@apache.org>.
tanmaya-panda1 commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1602156219
Thanks for your quick reply @exceptionfactory. Our primary focus was to develop the Azure Data Explorer sink for NiFi (PR in pipeline) and this Azure Data Explorer processor as a source for NiFi was done to complete the ecosystem involving Azure Data Explorer and NiFi. So we don't have any particular use case for Azure Data Explorer as a standalone source.
Maybe we can start with your suggestion to make our processor in line with other source processor and merge this PR. Once we get similar query on supporting standalone source for our processor, we can implement the same.
@exceptionfactory, what do you suggest on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tanmaya-panda1 commented on a diff in pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "tanmaya-panda1 (via GitHub)" <gi...@apache.org>.
tanmaya-panda1 commented on code in PR #7122:
URL: https://github.com/apache/nifi/pull/7122#discussion_r1209211426
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/QueryAzureDataExplorer.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParameter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = QueryAzureDataExplorer.ADX_QUERY_ERROR_MESSAGE, description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = QueryAzureDataExplorer.ADX_EXECUTED_QUERY, description = "Azure Data Explorer executed query.")
+})
+public class QueryAzureDataExplorer extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+
+ public static final Relationship SUCCESS = new Relationship.Builder()
+ .name("SUCCESS")
+ .description("Relationship for success")
+ .build();
+ public static final Relationship FAILED = new Relationship.Builder()
+ .name("FAILED")
+ .description("Relationship for failure")
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParameter.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParameter.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .build();
+ private final ObjectMapper objectMapper = new ObjectMapper();
+ private final Set<Relationship> relationships = Set.of(SUCCESS,FAILED);
+ private final List<PropertyDescriptor> descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ private AdxSourceConnectionService service;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ service = context.getProperty(ADX_SOURCE_SERVICE).asControllerService(AdxSourceConnectionService.class);
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile outgoingFlowFile;
+ String databaseName = context.getProperty(DB_NAME).getValue();
+ String adxQuery;
+ KustoQueryResponse kustoQueryResponse;
+
+ //checks if this processor has any preceding connection, if yes retrieve
+ if (context.hasIncomingConnection()) {
+ FlowFile incomingFlowFile = session.get();
+ //incoming connection exists but the incoming flowfile is null
+ if (incomingFlowFile == null && context.hasNonLoopConnection()) {
+ return;
+ }
+ //incoming connection exists and retrieve adxQuery from context
+ if (incomingFlowFile != null && incomingFlowFile.getSize() == 0) {
+ if (context.getProperty(ADX_QUERY).isSet()) {
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(incomingFlowFile).getValue();
+ } else {
+ String message = "FlowFile query is empty and no scheduled query is set";
+ getLogger().error(message);
+ incomingFlowFile = session.putAttribute(incomingFlowFile, ADX_QUERY_ERROR_MESSAGE, message);
+ session.transfer(incomingFlowFile, FAILED);
+ return;
+ }
+ } else {
+ try {
+ adxQuery = getQuery(session, incomingFlowFile);
+ } catch(IOException ioe) {
+ throw new ProcessException("Failed to read Query from FlowFile",ioe);
+ }
+ }
+ outgoingFlowFile = incomingFlowFile;
+ } else {
+ outgoingFlowFile = session.create();
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(outgoingFlowFile).getValue();
+ }
+
+ try {
+ //execute Query
+ kustoQueryResponse = executeQuery(databaseName,adxQuery);
+ if(!kustoQueryResponse.isError()){
+ try(ByteArrayInputStream bais = new ByteArrayInputStream(objectMapper.writeValueAsBytes(kustoQueryResponse.getTableData()))){
+ session.importFrom(bais, outgoingFlowFile);
+ }
Review Comment:
makes sense @exceptionfactory, fixed as advised
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #7122:
URL: https://github.com/apache/nifi/pull/7122#discussion_r1210276161
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/QueryAzureDataExplorer.java:
##########
@@ -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.
+ */
+package org.apache.nifi.processors.adx;
+
+import org.apache.nifi.adx.KustoQueryService;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParameter;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = QueryAzureDataExplorer.ADX_QUERY_ERROR_MESSAGE, description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = QueryAzureDataExplorer.ADX_EXECUTED_QUERY, description = "Azure Data Explorer executed query.")
+})
+public class QueryAzureDataExplorer extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+
+ public static final Relationship SUCCESS = new Relationship.Builder()
+ .name("SUCCESS")
+ .description("Relationship for success")
+ .build();
+ public static final Relationship FAILED = new Relationship.Builder()
+ .name("FAILED")
+ .description("Relationship for failure")
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParameter.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParameter.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(KustoQueryService.class)
+ .build();
+ private final Set<Relationship> relationships = Set.of(SUCCESS,FAILED);
+ private final List<PropertyDescriptor> descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ private KustoQueryService service;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ service = context.getProperty(ADX_SOURCE_SERVICE).asControllerService(KustoQueryService.class);
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile outgoingFlowFile;
+ String databaseName = context.getProperty(DB_NAME).getValue();
+ String adxQuery;
+ KustoQueryResponse kustoQueryResponse;
+
+ //checks if this processor has any preceding connection, if yes retrieve
+ if (context.hasIncomingConnection()) {
+ FlowFile incomingFlowFile = session.get();
+ //incoming connection exists but the incoming flowfile is null
+ if (incomingFlowFile == null && context.hasNonLoopConnection()) {
+ return;
+ }
+ //incoming connection exists and retrieve adxQuery from context
+ if (incomingFlowFile != null && incomingFlowFile.getSize() == 0) {
+ if (context.getProperty(ADX_QUERY).isSet()) {
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(incomingFlowFile).getValue();
+ } else {
+ String message = "FlowFile query is empty and no scheduled query is set";
+ getLogger().error(message);
+ incomingFlowFile = session.putAttribute(incomingFlowFile, ADX_QUERY_ERROR_MESSAGE, message);
+ session.transfer(incomingFlowFile, FAILED);
+ return;
+ }
+ } else {
+ try {
+ adxQuery = getQuery(session, incomingFlowFile);
+ } catch(IOException ioe) {
+ throw new ProcessException("Failed to read Query from FlowFile",ioe);
+ }
+ }
+ outgoingFlowFile = incomingFlowFile;
+ } else {
+ outgoingFlowFile = session.create();
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(outgoingFlowFile).getValue();
+ }
+
+ //execute Query
+ kustoQueryResponse = executeQuery(databaseName,adxQuery);
+ if(kustoQueryResponse.isError()){
+ //failure scenario
+ getLogger().error("Failed to execute query: {}", kustoQueryResponse.getErrorMessage());
+ outgoingFlowFile = session.putAttribute(outgoingFlowFile, ADX_QUERY_ERROR_MESSAGE, kustoQueryResponse.getErrorMessage());
+ session.transfer(outgoingFlowFile, FAILED);
+ }else {
+ // success scenario
+ session.importFrom(kustoQueryResponse.getAdxQueryResponseStream(), outgoingFlowFile);
Review Comment:
The `ProcessSession.importFrom()` method may not close the provided InputStream, so recommend wrapping this call in try-with-resources:
```suggestion
try (final InputStream responseStream = kustoQueryResponse.getAdxQueryResponseStream()) {
session.importFrom(responseStream, outgoingFlowFile);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #7122:
URL: https://github.com/apache/nifi/pull/7122#discussion_r1210277362
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/src/main/java/org/apache/nifi/adx/model/KustoQueryResponse.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.adx.model;
+
+import java.io.InputStream;
+
+public class KustoQueryResponse {
+
+ private InputStream adxQueryResponseStream;
+ private boolean isError;
Review Comment:
Minor naming adjustment recommendation:
```suggestion
private InputStream responseStream;
private boolean error;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #7122:
URL: https://github.com/apache/nifi/pull/7122#discussion_r1177940619
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/QueryAzureDataExplorer.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParameter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
Review Comment:
FlowFile attribute names generally follow the convention of lowercase with `.` separators. Recommend the following:
```suggestion
@WritesAttribute(attribute = "adx.query.error.message", description = "Azure Data Explorer error message."),
@WritesAttribute(attribute = "adx.query.executed", description = "Azure Data Explorer executed query.")
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/QueryAzureDataExplorer.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParameter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class QueryAzureDataExplorer extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
Review Comment:
These two values can be referenced in the `WritesAttribute` annotations.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/QueryAzureDataExplorer.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParameter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class QueryAzureDataExplorer extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final String RELATIONSHIP_SUCCESS = "SUCCESS";
+
+ public static final String RELATIONSHIP_FAILED = "FAILED";
+ public static final String RELATIONSHIP_FAILED_DESC = "Relationship for failure";
+ public static final String RELATIONSHIP_SUCCESS_DESC = "Relationship for success";
+
+ public static final Relationship SUCCESS = new Relationship.Builder()
+ .name(RELATIONSHIP_SUCCESS)
+ .description(RELATIONSHIP_SUCCESS_DESC)
+ .build();
+ public static final Relationship FAILED = new Relationship.Builder()
+ .name(RELATIONSHIP_FAILED)
+ .description(RELATIONSHIP_FAILED_DESC)
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParameter.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParameter.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .build();
+ private final ObjectMapper objectMapper = new ObjectMapper();
+ private Set<Relationship> relationships;
+ private List<PropertyDescriptor> descriptors;
+ private Client executionClient;
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ this.descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ this.relationships = Set.of(SUCCESS,FAILED);
Review Comment:
These collections can be declared statically and the `init` method can be removed.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,118 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
Review Comment:
This dependency is not necessary and should be removed.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,118 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
Review Comment:
Following general project conventions, the module name should include the word `service` indicating that it contains the Controller Service implementation.
```suggestion
<artifactId>nifi-adx-service</artifactId>
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.adx;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+
+import java.util.List;
+import static org.apache.nifi.adx.NiFiVersion.NIFI_SOURCE;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
Review Comment:
Controller Services do not read FlowFile attributes, so these annotations should be removed.
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.adx;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+
+import java.util.List;
+import static org.apache.nifi.adx.NiFiVersion.NIFI_SOURCE;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
+ adxConnectionParams = new ADXConnectionParams();
+ adxConnectionParams.setKustoAuthStrategy(context.getProperty(KUSTO_AUTH_STRATEGY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppId(context.getProperty(APP_ID).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppKey(context.getProperty(APP_KEY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppTenant(context.getProperty(APP_TENANT).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setKustoEngineURL(context.getProperty(CLUSTER_URL).evaluateAttributeExpressions().getValue());
+ if (this.executionClient != null) {
+ onStopped();
+ }
+
+ }
+
+ @OnStopped
+ public final void onStopped() {
+ if(this.executionClient!=null){
Review Comment:
```suggestion
if (this.executionClient != null) {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/NiFiVersion.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.adx;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NiFiVersion {
+ public static final String CLIENT_NAME = "Kusto.Nifi";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
Review Comment:
Is this value necessary?
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/test/java/org/apache/nifi/adx/MockAzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.adx;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class MockAzureAdxSourceProcessor extends AbstractProcessor {
Review Comment:
This class can be removed in favor of a mock created using Mockito.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/enums/AzureAdxSourceProcessorParameter.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx.enums;
+
+public enum AzureAdxSourceProcessorParameter {
+ DB_NAME("Database name", "The name of the database where the query will be executed."),
Review Comment:
Display Names should follow the "Title Case" convention, capitalizing most of the words.
```suggestion
DB_NAME("Database Name", "The name of the database where the query will be executed."),
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/QueryAzureDataExplorer.java:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParameter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class QueryAzureDataExplorer extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final String RELATIONSHIP_SUCCESS = "SUCCESS";
+
+ public static final String RELATIONSHIP_FAILED = "FAILED";
+ public static final String RELATIONSHIP_FAILED_DESC = "Relationship for failure";
+ public static final String RELATIONSHIP_SUCCESS_DESC = "Relationship for success";
Review Comment:
These descriptions should not be public static variables, instead they can be specified directly in the Relationship Builder.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/enums/AzureAdxSourceProcessorParameter.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx.enums;
+
+public enum AzureAdxSourceProcessorParameter {
+ DB_NAME("Database name", "The name of the database where the query will be executed."),
+
+ ADX_QUERY("ADX query", "The query which needs to be executed in Azure Data Explorer."),
Review Comment:
The `ADX` prefix on the property name is not necessary.
```suggestion
ADX_QUERY("Query", "The query which needs to be executed in Azure Data Explorer."),
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,118 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
Review Comment:
Compile dependencies should be listed before test dependencies.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.adx;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+
+import java.util.List;
+import static org.apache.nifi.adx.NiFiVersion.NIFI_SOURCE;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
Review Comment:
These static string values should be replaced with an `enum` that defines the allowed values. The `enum` should implement the `DescribedValue` interface to provide description information.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.adx;
Review Comment:
```suggestion
package org.apache.nifi.adx.service;
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.adx;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+
+import java.util.List;
+import static org.apache.nifi.adx.NiFiVersion.NIFI_SOURCE;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
+ adxConnectionParams = new ADXConnectionParams();
+ adxConnectionParams.setKustoAuthStrategy(context.getProperty(KUSTO_AUTH_STRATEGY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppId(context.getProperty(APP_ID).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppKey(context.getProperty(APP_KEY).evaluateAttributeExpressions().getValue());
Review Comment:
Sensitive properties should not support attribute evaluation.
```suggestion
adxConnectionParams.setAppKey(context.getProperty(APP_KEY).getValue());
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.adx;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+
+import java.util.List;
+import static org.apache.nifi.adx.NiFiVersion.NIFI_SOURCE;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
Review Comment:
This property should be marked as sensitive.
```suggestion
.required(true)
.sensitive(true)
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-nar/pom.xml:
##########
@@ -0,0 +1,63 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-nar</artifactId>
+ <packaging>nar</packaging>
+ <properties>
+ <maven.javadoc.skip>true</maven.javadoc.skip>
+ <source.skip>true</source.skip>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-processors</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <java.io.tmpdir>${project.build.directory}</java.io.tmpdir>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
+ </plugins>
Review Comment:
Is there a specific reason for changing the temporary directory? This seems unnecessary.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/src/main/java/org/apache/nifi/adx/AdxSourceConnectionService.java:
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.adx;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.controller.ControllerService;
+
+import com.microsoft.azure.kusto.data.Client;
Review Comment:
This `Client` references places a hard dependency on a specific version of the Kusto library. It looks like the use of the Client in the Processor is limited to return results. Instead of having the interface return the `Client`, it looks like this service interface could be abstract with a `query()` method that returns Data in the form of `List<List<Object>>`. What do you think of that approach? That would allow removing the dependency on the Kusto library from the API module, and make the service more flexible should it be necessary to upgrade the Kusto library for the Controller Service implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tanmaya-panda1 commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "tanmaya-panda1 (via GitHub)" <gi...@apache.org>.
tanmaya-panda1 commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1583953451
@exceptionfactory : A Gentle reminder to the PR, can you please review the changes once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #7122:
URL: https://github.com/apache/nifi/pull/7122#discussion_r1157598931
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/pom.xml:
##########
@@ -0,0 +1,89 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-api</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <java.io.tmpdir>${project.build.directory}</java.io.tmpdir>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
Review Comment:
This configuration should be removed. The standard temporary directory should be used without changing it to the project build directory.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/pom.xml:
##########
@@ -0,0 +1,89 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-api</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
Review Comment:
The code coverage plugin configuration should be removed. This is something that can be considered at a project level, but it should not be configured for specific modules.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/pom.xml:
##########
@@ -0,0 +1,89 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-api</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/pom.xml:
##########
@@ -0,0 +1,126 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx</artifactId>
+ <version>${nifi-release-version}</version>
Review Comment:
Using this `nifi-release-version` is not a standard across the repository. It should be replaced with the explicit version `2.0.0-SNAPSHOT`.
```suggestion
<version>2.0.0-SNAPSHOT</version>
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/pom.xml:
##########
@@ -0,0 +1,89 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-api</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <java.io.tmpdir>${project.build.directory}</java.io.tmpdir>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/pom.xml:
##########
@@ -0,0 +1,126 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
Review Comment:
These dependencies are applied at the project level and do not need to be declared for individual modules.
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/pom.xml:
##########
@@ -0,0 +1,126 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>5.9.0</version>
Review Comment:
The version for JUnit is set at the project level, so this explicit version should be removed.
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParamsEnum;
+import org.apache.nifi.processors.adx.enums.RelationshipStatusEnum;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class AzureAdxSourceProcessor extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final Relationship RL_SUCCEEDED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_SUCCEEDED.name())
+ .description(RelationshipStatusEnum.RL_SUCCEEDED.getDescription())
+ .build();
+ public static final Relationship RL_FAILED = new Relationship.Builder()
Review Comment:
```suggestion
public static final Relationship FAILED = new Relationship.Builder()
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParamsEnum;
+import org.apache.nifi.processors.adx.enums.RelationshipStatusEnum;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class AzureAdxSourceProcessor extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final Relationship RL_SUCCEEDED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_SUCCEEDED.name())
+ .description(RelationshipStatusEnum.RL_SUCCEEDED.getDescription())
+ .build();
+ public static final Relationship RL_FAILED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_FAILED.name())
+ .description(RelationshipStatusEnum.RL_FAILED.getDescription())
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .build();
+ private final ObjectMapper objectMapper = new ObjectMapper();
+ private Set<Relationship> relationships;
+ private List<PropertyDescriptor> descriptors;
+ private Client executionClient;
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ this.descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ this.relationships = Set.of(RL_SUCCEEDED,RL_FAILED);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ AdxSourceConnectionService service = context.getProperty(ADX_SOURCE_SERVICE).asControllerService(AdxSourceConnectionService.class);
+ executionClient = service.getKustoQueryClient();
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile outgoingFlowFile;
+ String databaseName = context.getProperty(DB_NAME).getValue();
+ String adxQuery;
+ KustoOperationResult kustoOperationResult;
+
+ //checks if this processor has any preceding connection, if yes retrieve
+ if ( context.hasIncomingConnection() ) {
Review Comment:
```suggestion
if (context.hasIncomingConnection()) {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParamsEnum;
+import org.apache.nifi.processors.adx.enums.RelationshipStatusEnum;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class AzureAdxSourceProcessor extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final Relationship RL_SUCCEEDED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_SUCCEEDED.name())
+ .description(RelationshipStatusEnum.RL_SUCCEEDED.getDescription())
+ .build();
+ public static final Relationship RL_FAILED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_FAILED.name())
+ .description(RelationshipStatusEnum.RL_FAILED.getDescription())
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .build();
+ private final ObjectMapper objectMapper = new ObjectMapper();
+ private Set<Relationship> relationships;
+ private List<PropertyDescriptor> descriptors;
+ private Client executionClient;
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ this.descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ this.relationships = Set.of(RL_SUCCEEDED,RL_FAILED);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ AdxSourceConnectionService service = context.getProperty(ADX_SOURCE_SERVICE).asControllerService(AdxSourceConnectionService.class);
+ executionClient = service.getKustoQueryClient();
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile outgoingFlowFile;
+ String databaseName = context.getProperty(DB_NAME).getValue();
+ String adxQuery;
+ KustoOperationResult kustoOperationResult;
+
+ //checks if this processor has any preceding connection, if yes retrieve
+ if ( context.hasIncomingConnection() ) {
+ FlowFile incomingFlowFile = session.get();
+ //incoming connection exists but the incoming flowfile is null
+ if ( incomingFlowFile == null && context.hasNonLoopConnection() ) {
+ getLogger().error("Incoming connection exists but Incoming flow file is null");
+ return;
+ }
+ //incoming connection exists and retrieve adxQuery from context
+ if ( incomingFlowFile != null && incomingFlowFile.getSize() == 0 ) {
+ if ( context.getProperty(ADX_QUERY).isSet() ) {
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(incomingFlowFile).getValue();
+ } else {
+ String message = "FlowFile query is empty and no scheduled query is set";
+ getLogger().error(message);
+ incomingFlowFile = session.putAttribute(incomingFlowFile, ADX_QUERY_ERROR_MESSAGE, message);
+ session.transfer(incomingFlowFile, RL_FAILED);
+ return;
+ }
+ } else {
+ try {
+ adxQuery = getQuery(session, incomingFlowFile);
+ } catch(IOException ioe) {
+ getLogger().error("Exception while reading from FlowFile " + ioe.getLocalizedMessage(), ioe);
+ throw new ProcessException(ioe);
+ }
+ }
+ outgoingFlowFile = incomingFlowFile;
+ } else {
+ outgoingFlowFile = session.create();
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(outgoingFlowFile).getValue();
+ }
+
+ try {
+ //execute Query
+ kustoOperationResult = executeQuery(databaseName,adxQuery);
+ List<List<Object>> tableData = kustoOperationResult.getPrimaryResults().getData();
+ String json = tableData.size() == 1 ? objectMapper.writeValueAsString(tableData.get(0)) : objectMapper.writeValueAsString(tableData);
+ if ( getLogger().isDebugEnabled() ) {
+ getLogger().debug("Query result {} ", tableData);
+ }
+ try(ByteArrayInputStream bais = new ByteArrayInputStream(json.getBytes())){
+ session.importFrom(bais, outgoingFlowFile);
+ }
Review Comment:
This type of approach should be changed to use stream the data to an OutputStream using ObjectMapper, instead of holding it all in memory.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,154 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
+ <version>1.15.3</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
Review Comment:
This explicit version and exclusion should be removed.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,154 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
+ <version>1.15.3</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.14.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-nar/pom.xml:
##########
@@ -0,0 +1,63 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-nar</artifactId>
+ <packaging>nar</packaging>
+ <properties>
+ <maven.javadoc.skip>true</maven.javadoc.skip>
+ <source.skip>true</source.skip>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-processors</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <java.io.tmpdir>${project.build.directory}</java.io.tmpdir>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
+ </plugins>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-nar/pom.xml:
##########
@@ -0,0 +1,63 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-nar</artifactId>
+ <packaging>nar</packaging>
+ <properties>
+ <maven.javadoc.skip>true</maven.javadoc.skip>
+ <source.skip>true</source.skip>
+ </properties>
Review Comment:
This options can be removed.
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/pom.xml:
##########
@@ -0,0 +1,126 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>5.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <version>5.9.0</version>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,154 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
+ <version>1.15.3</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.14.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>5.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>${nifi-release-version}</version>
+ <type>nar</type>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>compile</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <java.io.tmpdir>${project.build.directory}</java.io.tmpdir>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/pom.xml:
##########
@@ -0,0 +1,126 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>5.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <version>5.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxConnectionServiceParamsEnum.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.adx;
+
+public enum AzureAdxConnectionServiceParamsEnum {
Review Comment:
```suggestion
public enum AzureAdxConnectionServiceParameter {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/NiFiVersion.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.adx;
+
+import org.apache.commons.lang3.tuple.Pair;import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.InputStream;
+import java.util.Properties;
+
+public class NiFiVersion {
+ public static final String CLIENT_NAME = "Kusto.Nifi";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+ private static final Logger log = LoggerFactory.getLogger(NiFiVersion.class);
+ private static final String VERSION_FILE = "/azure-kusto-nifi-version.properties";
+ private static String version = "unknown";
+
+ private NiFiVersion() {
+ }
+
+ static {
+ try {
+ Properties props = new Properties();
+ try (InputStream versionFileStream = NiFiVersion.class.getResourceAsStream(VERSION_FILE)) {
+ props.load(versionFileStream);
+ version = props.getProperty("version", version).trim();
+ }
+ } catch (Exception e) {
+ log.warn("Error while loading version:", e);
+ }
Review Comment:
This approach should be refactored to use `getPackage().getImplementationVersion()` along with adding the information to the module manifest configuration.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParamsEnum;
+import org.apache.nifi.processors.adx.enums.RelationshipStatusEnum;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class AzureAdxSourceProcessor extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final Relationship RL_SUCCEEDED = new Relationship.Builder()
Review Comment:
Using the `RL` prefix is not a standard convention. Although `REL` is sometimes used, recommend eliminating the prefix and changing the name to `SUCCESS`.
```suggestion
public static final Relationship SUCCESS = new Relationship.Builder()
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/resources/azure-kusto-nifi-version.properties:
##########
@@ -0,0 +1,17 @@
+#
+# 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.
+#
+version=${project.version}
Review Comment:
As mentioned in the `NiFiVersion` class, this approach should be changed to use the standard Java `getPackage().getImplementationVersion()` approach based on module manifest information.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/pom.xml:
##########
@@ -0,0 +1,126 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>5.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <version>5.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <java.io.tmpdir>${project.build.directory}</java.io.tmpdir>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/enums/RelationshipStatusEnum.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx.enums;
+
+public enum RelationshipStatusEnum {
Review Comment:
This enum is probably not necessary as the values can be declared directly in the Relationship definition.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/enums/AzureAdxSourceProcessorParamsEnum.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx.enums;
+
+public enum AzureAdxSourceProcessorParamsEnum {
Review Comment:
The `Enum` suffix should be avoided.
```suggestion
public enum AzureAdxSourceProcessorParameter {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/NiFiVersion.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.adx;
+
+import org.apache.commons.lang3.tuple.Pair;import org.slf4j.Logger;
Review Comment:
```suggestion
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParamsEnum;
+import org.apache.nifi.processors.adx.enums.RelationshipStatusEnum;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class AzureAdxSourceProcessor extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final Relationship RL_SUCCEEDED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_SUCCEEDED.name())
+ .description(RelationshipStatusEnum.RL_SUCCEEDED.getDescription())
+ .build();
+ public static final Relationship RL_FAILED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_FAILED.name())
+ .description(RelationshipStatusEnum.RL_FAILED.getDescription())
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .build();
+ private final ObjectMapper objectMapper = new ObjectMapper();
+ private Set<Relationship> relationships;
+ private List<PropertyDescriptor> descriptors;
+ private Client executionClient;
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ this.descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ this.relationships = Set.of(RL_SUCCEEDED,RL_FAILED);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ AdxSourceConnectionService service = context.getProperty(ADX_SOURCE_SERVICE).asControllerService(AdxSourceConnectionService.class);
+ executionClient = service.getKustoQueryClient();
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile outgoingFlowFile;
+ String databaseName = context.getProperty(DB_NAME).getValue();
+ String adxQuery;
+ KustoOperationResult kustoOperationResult;
+
+ //checks if this processor has any preceding connection, if yes retrieve
+ if ( context.hasIncomingConnection() ) {
+ FlowFile incomingFlowFile = session.get();
+ //incoming connection exists but the incoming flowfile is null
+ if ( incomingFlowFile == null && context.hasNonLoopConnection() ) {
+ getLogger().error("Incoming connection exists but Incoming flow file is null");
+ return;
+ }
+ //incoming connection exists and retrieve adxQuery from context
+ if ( incomingFlowFile != null && incomingFlowFile.getSize() == 0 ) {
+ if ( context.getProperty(ADX_QUERY).isSet() ) {
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(incomingFlowFile).getValue();
+ } else {
+ String message = "FlowFile query is empty and no scheduled query is set";
+ getLogger().error(message);
+ incomingFlowFile = session.putAttribute(incomingFlowFile, ADX_QUERY_ERROR_MESSAGE, message);
+ session.transfer(incomingFlowFile, RL_FAILED);
+ return;
+ }
+ } else {
+ try {
+ adxQuery = getQuery(session, incomingFlowFile);
+ } catch(IOException ioe) {
+ getLogger().error("Exception while reading from FlowFile " + ioe.getLocalizedMessage(), ioe);
+ throw new ProcessException(ioe);
Review Comment:
It is not necessary to log an error when throwing an exception, since the framework will log exception as an error. Also recommend making the error message more descriptive.
```suggestion
throw new ProcessException("Failed to read Query from FlowFile", ioe);
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParamsEnum;
+import org.apache.nifi.processors.adx.enums.RelationshipStatusEnum;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class AzureAdxSourceProcessor extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final Relationship RL_SUCCEEDED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_SUCCEEDED.name())
+ .description(RelationshipStatusEnum.RL_SUCCEEDED.getDescription())
+ .build();
+ public static final Relationship RL_FAILED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_FAILED.name())
+ .description(RelationshipStatusEnum.RL_FAILED.getDescription())
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .build();
+ private final ObjectMapper objectMapper = new ObjectMapper();
+ private Set<Relationship> relationships;
+ private List<PropertyDescriptor> descriptors;
+ private Client executionClient;
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ this.descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ this.relationships = Set.of(RL_SUCCEEDED,RL_FAILED);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ AdxSourceConnectionService service = context.getProperty(ADX_SOURCE_SERVICE).asControllerService(AdxSourceConnectionService.class);
+ executionClient = service.getKustoQueryClient();
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile outgoingFlowFile;
+ String databaseName = context.getProperty(DB_NAME).getValue();
+ String adxQuery;
+ KustoOperationResult kustoOperationResult;
+
+ //checks if this processor has any preceding connection, if yes retrieve
+ if ( context.hasIncomingConnection() ) {
+ FlowFile incomingFlowFile = session.get();
+ //incoming connection exists but the incoming flowfile is null
+ if ( incomingFlowFile == null && context.hasNonLoopConnection() ) {
+ getLogger().error("Incoming connection exists but Incoming flow file is null");
+ return;
+ }
+ //incoming connection exists and retrieve adxQuery from context
+ if ( incomingFlowFile != null && incomingFlowFile.getSize() == 0 ) {
+ if ( context.getProperty(ADX_QUERY).isSet() ) {
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(incomingFlowFile).getValue();
+ } else {
+ String message = "FlowFile query is empty and no scheduled query is set";
+ getLogger().error(message);
+ incomingFlowFile = session.putAttribute(incomingFlowFile, ADX_QUERY_ERROR_MESSAGE, message);
+ session.transfer(incomingFlowFile, RL_FAILED);
+ return;
+ }
+ } else {
+ try {
+ adxQuery = getQuery(session, incomingFlowFile);
+ } catch(IOException ioe) {
+ getLogger().error("Exception while reading from FlowFile " + ioe.getLocalizedMessage(), ioe);
+ throw new ProcessException(ioe);
+ }
+ }
+ outgoingFlowFile = incomingFlowFile;
+ } else {
+ outgoingFlowFile = session.create();
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(outgoingFlowFile).getValue();
+ }
+
+ try {
+ //execute Query
+ kustoOperationResult = executeQuery(databaseName,adxQuery);
+ List<List<Object>> tableData = kustoOperationResult.getPrimaryResults().getData();
+ String json = tableData.size() == 1 ? objectMapper.writeValueAsString(tableData.get(0)) : objectMapper.writeValueAsString(tableData);
+ if ( getLogger().isDebugEnabled() ) {
+ getLogger().debug("Query result {} ", tableData);
+ }
Review Comment:
The conditional wrapper is not necessary since logging a message with a placeholder does not involve significant additional work. In addition, entire data content should never be logged since it is available through standard provenance features. Recommend removing this debugging.
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/pom.xml:
##########
@@ -0,0 +1,67 @@
+<?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-nar-bundles</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <packaging>pom</packaging>
+
+ <properties>
+ <nifi-release-version>2.0.0-SNAPSHOT</nifi-release-version>
+ <azure-kusto-java-sdk-version>4.0.4</azure-kusto-java-sdk-version>
+ </properties>
+
+
+ <modules>
+ <module>nifi-adx-processors</module>
+ <module>nifi-adx-nar</module>
+ <module>nifi-adx</module>
+ <module>nifi-adx-api</module>
+ </modules>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParamsEnum;
+import org.apache.nifi.processors.adx.enums.RelationshipStatusEnum;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class AzureAdxSourceProcessor extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+ public static final Relationship RL_SUCCEEDED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_SUCCEEDED.name())
+ .description(RelationshipStatusEnum.RL_SUCCEEDED.getDescription())
+ .build();
+ public static final Relationship RL_FAILED = new Relationship.Builder()
+ .name(RelationshipStatusEnum.RL_FAILED.name())
+ .description(RelationshipStatusEnum.RL_FAILED.getDescription())
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParamsEnum.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .build();
+ private final ObjectMapper objectMapper = new ObjectMapper();
+ private Set<Relationship> relationships;
+ private List<PropertyDescriptor> descriptors;
+ private Client executionClient;
+
+ @Override
+ protected void init(final ProcessorInitializationContext context) {
+ this.descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ this.relationships = Set.of(RL_SUCCEEDED,RL_FAILED);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ AdxSourceConnectionService service = context.getProperty(ADX_SOURCE_SERVICE).asControllerService(AdxSourceConnectionService.class);
+ executionClient = service.getKustoQueryClient();
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile outgoingFlowFile;
+ String databaseName = context.getProperty(DB_NAME).getValue();
+ String adxQuery;
+ KustoOperationResult kustoOperationResult;
+
+ //checks if this processor has any preceding connection, if yes retrieve
+ if ( context.hasIncomingConnection() ) {
+ FlowFile incomingFlowFile = session.get();
+ //incoming connection exists but the incoming flowfile is null
+ if ( incomingFlowFile == null && context.hasNonLoopConnection() ) {
+ getLogger().error("Incoming connection exists but Incoming flow file is null");
Review Comment:
This status should not be logged as an error, it is a framework concern, so recommend changing the log to an info or removing the log.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/test/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessorE2ETest.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import org.apache.nifi.adx.AzureAdxSourceConnectionService;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+
+/**
+ * These are end to end tests for ADX Source Processor, which are disabled/skipped by default
+ * during maven test phase. These parameters need to be provided during maven build
+ * -DexecuteE2ETests=(set to true)
+ * -DappId=(appId)
+ * -DappKey=(appKey)
+ * -DappTenant=(appTenant)
+ * -DclusterUrl=(clusterUrl)
+ * -DdatabaseName=(databaseName)
+ * -DadxQuery=(query-to-be-executed-in-source-ADX)
+ * -DadxQueryWhichExceedsLimit=(query-who-execution-will-exceed-the-kusto-query-limits)
+ **/
+class AzureAdxSourceProcessorE2ETest {
+
+ private TestRunner testRunner;
+
+ @BeforeEach
+ public void init() throws InitializationException {
+ Assumptions.assumeTrue("true".equalsIgnoreCase(System.getProperty("executeE2ETests")));
Review Comment:
This approach should be changed to use the `EnabledIfSystemProperty` annotation
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,154 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
+ <version>1.15.3</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.14.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>5.9.0</version>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,154 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
+ <version>1.15.3</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.14.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>5.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>${nifi-release-version}</version>
+ <type>nar</type>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>${nifi-release-version}</version>
+ <scope>compile</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.6</version>
+ <executions>
+ <execution>
+ <id>prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,154 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
+ <version>1.15.3</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.14.0</version>
Review Comment:
This version is managed at the root level and should be removed.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,154 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>${nifi-release-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-processor-utils</artifactId>
Review Comment:
The `nifi-processor-utils` library is no longer part of the project, so this dependency needs to be replaced with the specific dependency required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tanmaya-panda1 commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "tanmaya-panda1 (via GitHub)" <gi...@apache.org>.
tanmaya-panda1 commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1567037157
@exceptionfactory, Thanks for your review comments, I have fixed and raised the PR. Also I have removed the testcase for kusto query limits exceeded, since the query response limits the results to 500,000 rows when the number of records exceeds more than 500,000 rows. Re-requesting your review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tanmaya-panda1 commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "tanmaya-panda1 (via GitHub)" <gi...@apache.org>.
tanmaya-panda1 commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1576113721
@exceptionfactory : Thanks for reviewing. I have done the changes as requested, can you please have a look at the PR again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #7122: NIFI-8294 Add Query Processor for Azure Data Explorer
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1602561371
Thanks for the reply @tanmaya-panda1. With that background, it sounds like merging this PR will provide a helpful baseline for connectivity to Azure Data Explorer. Of course future changes could be implemented as needed. Will plan on merging soon!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] kustonaut commented on pull request #7122: NIFI-8294 Add Query Processor for Azure Data Explorer
Posted by "kustonaut (via GitHub)" <gi...@apache.org>.
kustonaut commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1603587094
[celebrate] Akshay Dixit reacted to your message:
________________________________
From: exceptionfactory ***@***.***>
Sent: Thursday, June 22, 2023 8:15:59 PM
To: apache/nifi ***@***.***>
Cc: Akshay Dixit ***@***.***>; Mention ***@***.***>
Subject: Re: [apache/nifi] NIFI-8294 Add Query Processor for Azure Data Explorer (PR #7122)
@exceptionfactory approved this pull request.
—
Reply to this email directly, view it on GitHub<https://github.com/apache/nifi/pull/7122#pullrequestreview-1493931033>, or unsubscribe<https://github.com/notifications/unsubscribe-auth/ATWGSJDCTOMV6ZNDK3YOW5DXMSRX7ANCNFSM6AAAAAAWS4NZRM>.
You are receiving this because you were mentioned.Message ID: ***@***.***>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1600758257
> Thanks a ton @exceptionfactory for reviewing the PR and implementing the fixes. I think we are good with all changes as suggested. But I had a question though on QueryAzureDataExplorer. Since it mandatorily requires a flow file to query into ADX, can this processor become a starting point of a flow where we query into ADX using input query and write the output to a file or some other output processor?
You're welcome, and thanks for the follow-up @tanmaya-panda1.
Regarding the required input FlowFile, you are right that the updated approach changes the behavior to require some other starting point Processor, such as GenerateFlowFile, as an input to QueryAzureDataExplorer. This behavior could be restored, but I wanted to clarify the intended use case first. Other "source" Processors, such as the List Processors, use state tracking to provide a query approach that only brings in new information. Without that tracking in this Processor, it would be necessary to right a query with a relative boundary, or using NiFi Expression Language to compute the current time, for example. The same result could be achieved using `GenerateFlowFile` in front of this Processor, but I understand that it introduces an additional step in the data flow. If you can provide some additional details on how you expected to use this Processor as a standalone source, that would be helpful in determining the best path forward.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] kustonaut commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "kustonaut (via GitHub)" <gi...@apache.org>.
kustonaut commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1558598449
Hi @exceptionfactory,
Requesting your re-review as we have addressed your comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1560280573
> Hi @exceptionfactory, Requesting your re-review as we have addressed your comments.
Thanks @kustonaut and @tanmaya-panda1, I will follow up on this soon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #7122:
URL: https://github.com/apache/nifi/pull/7122#discussion_r1203243558
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/pom.xml:
##########
@@ -0,0 +1,49 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-api</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
Review Comment:
This block is not necessary since the `nifi-adx-api` module does not contain any test sources.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-nar/pom.xml:
##########
@@ -0,0 +1,57 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-nar</artifactId>
+ <packaging>nar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-processors</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
Review Comment:
This block is not necessary because `nifi-adx-nar` does not contain any tests or test resources.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/src/main/java/org/apache/nifi/adx/AdxSourceConnectionService.java:
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.adx;
+
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.controller.ControllerService;
+
+@Tags({"azure", "adx"})
+@CapabilityDescription("Connection-Service to Azure ADX (Kusto) cluster.")
+public interface AdxSourceConnectionService extends ControllerService {
Review Comment:
Recommend renaming this service interface.
```suggestion
public interface KustoQueryService extends ControllerService {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,115 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
Review Comment:
The license plugin execution cannot be skipped as part of standard execution, this plugin configuration should be removed.
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxConnectionServiceParameter.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.adx;
+
+public enum AzureAdxConnectionServiceParameter {
+
+ AUTH_STRATEGY("Kusto Authentication Method", "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
Review Comment:
```suggestion
AUTHENTICATION_METHOD("Kusto Authentication Method", "The strategy or method to authenticate against Azure Data Explorer"),
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
Review Comment:
This log message duplicates standard framework logging and is not necessary.
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
Review Comment:
```suggestion
private Client kustoClient;
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
+ adxConnectionParams = new ADXConnectionParams();
+ adxConnectionParams.setKustoAuthStrategy(context.getProperty(KUSTO_AUTH_STRATEGY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppId(context.getProperty(APP_ID).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppKey(context.getProperty(APP_KEY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppTenant(context.getProperty(APP_TENANT).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setKustoEngineURL(context.getProperty(CLUSTER_URL).evaluateAttributeExpressions().getValue());
+ if (this.executionClient != null) {
+ onStopped();
+ }
+
+ }
+
+ @OnStopped
+ public final void onStopped() {
+ if(this.executionClient!=null){
+ try {
+ this.executionClient.close();
+ } catch (Exception e) {
+ getLogger().error("Error closing Kusto Execution Client", e);
+ }
+ }
+ this.executionClient = null;
+ }
+
+
+ public Client getKustoQueryClient() {
+ return createKustoExecutionClient(adxConnectionParams.getKustoEngineURL(),
+ adxConnectionParams.getAppId(),
+ adxConnectionParams.getAppKey(),
+ adxConnectionParams.getAppTenant(),
+ adxConnectionParams.getKustoAuthStrategy());
+ }
+
+ @Override
+ public KustoQueryResponse executeQuery(String databaseName, String query){
+ if (this.executionClient == null) {
+ this.executionClient = getKustoQueryClient();
+ }
Review Comment:
This approach to initialization is not thread-safe. It looks like creating the Kusto Client could be moved to `onEnabled()`.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-nar/pom.xml:
##########
@@ -0,0 +1,57 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-nar</artifactId>
+ <packaging>nar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-processors</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/resources/azure-kusto-nifi-version.properties:
##########
@@ -0,0 +1,17 @@
+#
+# 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.
+#
+version=${project.version}
Review Comment:
This file can be removed
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/test/java/org/apache/nifi/adx/TestAzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.adx;
+
+import com.microsoft.azure.kusto.data.Client;
+import org.apache.nifi.adx.service.AzureAdxSourceConnectionService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.ControllerServiceConfiguration;
+import org.apache.nifi.util.StandardProcessorTestRunner;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestAzureAdxSourceConnectionService {
+
+ private TestRunner runner;
+
+ private AzureAdxSourceConnectionService service;
+
+ private static final String MOCK_APP_ID = "mockAppId";
+
+ private static final String MOCK_APP_KEY = "mockAppKey";
+
+ private static final String MOCK_APP_TENANT = "mockAppTenant";
+
+ private static final String MOCK_CLUSTER_URL = "https://mockClusterUrl.com/";
+
+ @BeforeEach
+ public void setup() throws InitializationException {
+ runner = TestRunners.newTestRunner(new AbstractProcessor() {
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ List<PropertyDescriptor> propDescs = new ArrayList<>();
+ propDescs.add(new PropertyDescriptor.Builder()
+ .name("AdxService")
+ .description("AdxService")
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .required(true)
+ .build());
+ return propDescs;
+ }
+ });
+
+ service = new AzureAdxSourceConnectionService();
+ runner.addControllerService("test-good", service);
+ }
+
+ @AfterEach
+ public void after() {
+ runner.clearProperties();
+ }
+
+ /**
+ * test successful adx connection scenario where all valid parameters are passed
+ */
+ @Test
+ void testAdxConnectionController() {
+ configureAppId();
+ configureAppKey();
+ configureAppTenant();
+ configureClusterURL();
+
+ runner.assertValid(service);
+ }
+
+ /**
+ * test successful adx connection scenario where all valid parameters are passed
+ */
+ @Test
+ void testCreateExecutionClientSuccess() {
+ configureAppId();
+ configureAppKey();
+ configureAppTenant();
+ configureClusterURL();
+ runner.assertValid(service);
+ runner.setValidateExpressionUsage(false);
+ runner.enableControllerService(service);
+ Client executionClient = service.getKustoQueryClient();
+
+ Assertions.assertNotNull(executionClient);
+ }
+
+ @Test
+ void testPropertyDescriptor() {
+ configureAppId();
+ configureAppKey();
+ configureAppTenant();
+ configureClusterURL();
+ List<PropertyDescriptor> pd = service.getSupportedPropertyDescriptors();
+
+ assertTrue(pd.contains(AzureAdxSourceConnectionService.APP_ID));
+ assertTrue(pd.contains(AzureAdxSourceConnectionService.APP_KEY));
+ assertTrue(pd.contains(AzureAdxSourceConnectionService.APP_TENANT));
+ assertTrue(pd.contains(AzureAdxSourceConnectionService.CLUSTER_URL));
+ }
+
+
+ @Test
+ void testInvalidConnectionMissingProperty() {
+ configureAppId();
+ configureAppKey();
+ configureAppTenant();
+ runner.assertNotValid(service);
+ runner.setValidateExpressionUsage(false);
+ Assertions.assertNull(
+ ((ControllerServiceConfiguration)
+ ((Map.Entry<?, ?>) (((StandardProcessorTestRunner)runner).getProcessContext().getControllerServices()).entrySet().toArray()[0])
+ .getValue()).getProperty(AzureAdxSourceConnectionService.CLUSTER_URL));
Review Comment:
This deep nesting should be refactored to multiple interim declared variables.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxConnectionServiceParameter.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.adx;
+
+public enum AzureAdxConnectionServiceParameter {
Review Comment:
Recommend shortening this name.
```suggestion
public enum AzureDataExplorerParameter {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/pom.xml:
##########
@@ -0,0 +1,49 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx-api</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/resources/*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
Review Comment:
```suggestion
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-api/src/main/java/org/apache/nifi/adx/AdxSourceConnectionService.java:
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.adx;
+
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.controller.ControllerService;
+
+@Tags({"azure", "adx"})
Review Comment:
```suggestion
@Tags({"azure", "adx", "explorer", "kusto"})
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/pom.xml:
##########
@@ -0,0 +1,115 @@
+<?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-adx-bundle</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-adx</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-adx-api</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <type>nar</type>
+ </dependency>
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${azure-kusto-java-sdk-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <version>3.1.2</version>
+ <configuration>
+ <archive>
+ <manifest>
+ <packageName>org.apache.nifi.adx</packageName>
+ </manifest>
+ <manifestEntries>
+ <implementation-title>nifi-adx</implementation-title>
+ <implementation-version>${project.version}</implementation-version>
+ </manifestEntries>
Review Comment:
Instead of using these custom entries, the standard `addDefaultImplementationEntries` element should be set in the `manifest` section.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/NiFiVersion.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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.adx;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NiFiVersion {
+ public static final String CLIENT_NAME = "Kusto.Nifi";
+ private static final Logger log = LoggerFactory.getLogger(NiFiVersion.class);
+ private static String version = "unknown";
+
+ private NiFiVersion() {
+ }
+
+ static {
+ try {
+ version = NiFiVersion.class.getPackage().getImplementationVersion();
+ log.info("Loaded version: {}", version);
+ } catch (Exception e) {
+ log.warn("Error while loading version:", e);
Review Comment:
This logging is not necessary and should be removed.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/model/ADXConnectionParams.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.adx.model;
+
+public class ADXConnectionParams {
Review Comment:
```suggestion
public class AzureDataExplorerConnectionParameters {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
Review Comment:
Following project conventions, recommend the following name aligned with the suggestion for the interface name.
```suggestion
public class StandardKustoQueryService extends AbstractControllerService implements AdxSourceConnectionService {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/AzureAdxConnectionServiceParameter.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.adx;
+
+public enum AzureAdxConnectionServiceParameter {
+
+ AUTH_STRATEGY("Kusto Authentication Method", "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ APP_ID("Application ID", "Azure application ID for accessing the ADX-Cluster"),
+ APP_KEY("Application KEY", "Azure application Key for accessing the ADX-Cluster"),
+ APP_TENANT("Application Tenant", "Azure application tenant for accessing the ADX-Cluster"),
+ CLUSTER_URL("Cluster URL", "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled.");
+
+ private final String paramDisplayName;
+ private final String description;
+
+
+ AzureAdxConnectionServiceParameter(String paramDisplayName, String description) {
+ this.paramDisplayName = paramDisplayName;
+ this.description = description;
+ }
+
+ public String getParamDisplayName() {
Review Comment:
```suggestion
public String getDisplayName() {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
+ adxConnectionParams = new ADXConnectionParams();
+ adxConnectionParams.setKustoAuthStrategy(context.getProperty(KUSTO_AUTH_STRATEGY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppId(context.getProperty(APP_ID).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppKey(context.getProperty(APP_KEY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppTenant(context.getProperty(APP_TENANT).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setKustoEngineURL(context.getProperty(CLUSTER_URL).evaluateAttributeExpressions().getValue());
+ if (this.executionClient != null) {
+ onStopped();
+ }
+
+ }
+
+ @OnStopped
+ public final void onStopped() {
+ if(this.executionClient!=null){
+ try {
+ this.executionClient.close();
+ } catch (Exception e) {
+ getLogger().error("Error closing Kusto Execution Client", e);
Review Comment:
```suggestion
getLogger().error("Kusto Client close failed", e);
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
+ adxConnectionParams = new ADXConnectionParams();
+ adxConnectionParams.setKustoAuthStrategy(context.getProperty(KUSTO_AUTH_STRATEGY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppId(context.getProperty(APP_ID).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppKey(context.getProperty(APP_KEY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppTenant(context.getProperty(APP_TENANT).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setKustoEngineURL(context.getProperty(CLUSTER_URL).evaluateAttributeExpressions().getValue());
Review Comment:
The properties do not declare support for Expression Language, which they should not, but the `evaluateAttributeExpression()` calls should be removed.
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
+ adxConnectionParams = new ADXConnectionParams();
+ adxConnectionParams.setKustoAuthStrategy(context.getProperty(KUSTO_AUTH_STRATEGY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppId(context.getProperty(APP_ID).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppKey(context.getProperty(APP_KEY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppTenant(context.getProperty(APP_TENANT).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setKustoEngineURL(context.getProperty(CLUSTER_URL).evaluateAttributeExpressions().getValue());
+ if (this.executionClient != null) {
+ onStopped();
+ }
+
+ }
+
+ @OnStopped
+ public final void onStopped() {
+ if(this.executionClient!=null){
+ try {
+ this.executionClient.close();
+ } catch (Exception e) {
+ getLogger().error("Error closing Kusto Execution Client", e);
+ }
+ }
+ this.executionClient = null;
+ }
+
+
+ public Client getKustoQueryClient() {
+ return createKustoExecutionClient(adxConnectionParams.getKustoEngineURL(),
+ adxConnectionParams.getAppId(),
+ adxConnectionParams.getAppKey(),
+ adxConnectionParams.getAppTenant(),
+ adxConnectionParams.getKustoAuthStrategy());
+ }
+
+ @Override
+ public KustoQueryResponse executeQuery(String databaseName, String query){
+ if (this.executionClient == null) {
+ this.executionClient = getKustoQueryClient();
+ }
+ List<List<Object>> tableData;
+ KustoQueryResponse kustoQueryResponse;
+ KustoOperationResult kustoOperationResult;
+ try {
+ kustoOperationResult = this.executionClient.execute(databaseName, query);
+ } catch (DataServiceException | DataClientException e) {
+ String errorMessage;
+ if(Arrays.stream(ExceptionUtils.getRootCauseStackTrace(e)).anyMatch(str -> str.contains("LimitsExceeded"))){
+ errorMessage = "Exception occurred while reading data from ADX : Query Limits exceeded : Please modify your query to fetch results below the kusto query limits";
Review Comment:
This message can be shortened:
```suggestion
errorMessage = "Query Limits exceeded: Please modify the query to fetch results below Kusto limits";
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
+ adxConnectionParams = new ADXConnectionParams();
+ adxConnectionParams.setKustoAuthStrategy(context.getProperty(KUSTO_AUTH_STRATEGY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppId(context.getProperty(APP_ID).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppKey(context.getProperty(APP_KEY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppTenant(context.getProperty(APP_TENANT).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setKustoEngineURL(context.getProperty(CLUSTER_URL).evaluateAttributeExpressions().getValue());
+ if (this.executionClient != null) {
+ onStopped();
+ }
+
+ }
+
+ @OnStopped
+ public final void onStopped() {
+ if(this.executionClient!=null){
+ try {
+ this.executionClient.close();
+ } catch (Exception e) {
+ getLogger().error("Error closing Kusto Execution Client", e);
+ }
+ }
+ this.executionClient = null;
+ }
+
+
+ public Client getKustoQueryClient() {
+ return createKustoExecutionClient(adxConnectionParams.getKustoEngineURL(),
+ adxConnectionParams.getAppId(),
+ adxConnectionParams.getAppKey(),
+ adxConnectionParams.getAppTenant(),
+ adxConnectionParams.getKustoAuthStrategy());
+ }
+
+ @Override
+ public KustoQueryResponse executeQuery(String databaseName, String query){
+ if (this.executionClient == null) {
+ this.executionClient = getKustoQueryClient();
+ }
+ List<List<Object>> tableData;
+ KustoQueryResponse kustoQueryResponse;
+ KustoOperationResult kustoOperationResult;
+ try {
+ kustoOperationResult = this.executionClient.execute(databaseName, query);
+ } catch (DataServiceException | DataClientException e) {
+ String errorMessage;
+ if(Arrays.stream(ExceptionUtils.getRootCauseStackTrace(e)).anyMatch(str -> str.contains("LimitsExceeded"))){
+ errorMessage = "Exception occurred while reading data from ADX : Query Limits exceeded : Please modify your query to fetch results below the kusto query limits";
+ }else {
+ errorMessage = "Exception occurred while reading data from ADX";
+ }
+ getLogger().error(errorMessage, e);
+ kustoQueryResponse = new KustoQueryResponse(true,errorMessage);
Review Comment:
```suggestion
kustoQueryResponse = new KustoQueryResponse(true, errorMessage);
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/main/java/org/apache/nifi/adx/service/AzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.adx.service;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.AzureAdxConnectionServiceParameter;
+import org.apache.nifi.adx.NiFiVersion;
+import org.apache.nifi.adx.model.ADXConnectionParams;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.util.Arrays;
+import java.util.List;
+
+@Tags({"Azure", "ADX", "Kusto", "ingest", "azure"})
+@CapabilityDescription("Sends batches of flow file content or stream flow file content to an Azure ADX cluster.")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "AUTH_STRATEGY", description = "The strategy/method to authenticate against Azure Active Directory, either 'application' or 'managed_identity'."),
+ @ReadsAttribute(attribute = "APP_ID", description = "Specifies Azure application id for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_KEY", description = "Specifies Azure application key for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "APP_TENANT", description = "Azure application tenant for accessing the ADX-Cluster."),
+ @ReadsAttribute(attribute = "CLUSTER_URL", description = "Endpoint of ADX cluster. This is required only when streaming data to ADX cluster is enabled."),
+})
+public class AzureAdxSourceConnectionService extends AbstractControllerService implements AdxSourceConnectionService {
+
+ private static final String KUSTO_STRATEGY_APPLICATION = "application";
+
+ private static final String KUSTO_STRATEGY_MANAGED_IDENTITY = "managed_identity";
+
+ public static final Pair<String,String> NIFI_SOURCE = Pair.of("processor", "nifi-source");
+
+ public static final PropertyDescriptor KUSTO_AUTH_STRATEGY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.name())
+ .displayName(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.AUTH_STRATEGY.getDescription())
+ .required(false)
+ .defaultValue(KUSTO_STRATEGY_APPLICATION)
+ .allowableValues(KUSTO_STRATEGY_APPLICATION, KUSTO_STRATEGY_MANAGED_IDENTITY)
+ .build();
+
+ public static final PropertyDescriptor APP_ID = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_ID.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_ID.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_ID.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_KEY = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_KEY.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_KEY.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_KEY.getDescription())
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor APP_TENANT = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.APP_TENANT.name())
+ .displayName(AzureAdxConnectionServiceParameter.APP_TENANT.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.APP_TENANT.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLUSTER_URL = new PropertyDescriptor
+ .Builder().name(AzureAdxConnectionServiceParameter.CLUSTER_URL.name())
+ .displayName(AzureAdxConnectionServiceParameter.CLUSTER_URL.getParamDisplayName())
+ .description(AzureAdxConnectionServiceParameter.CLUSTER_URL.getDescription())
+ .required(true)
+ .addValidator(StandardValidators.URL_VALIDATOR)
+ .build();
+
+ private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of(KUSTO_AUTH_STRATEGY,APP_ID,APP_KEY,APP_TENANT,CLUSTER_URL);
+
+ private Client executionClient;
+
+ private ADXConnectionParams adxConnectionParams;
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTY_DESCRIPTORS;
+ }
+
+ /**
+ * @param context the configuration context
+ */
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws ProcessException {
+ getLogger().info("Starting Azure ADX Source Connection Service...");
+ adxConnectionParams = new ADXConnectionParams();
+ adxConnectionParams.setKustoAuthStrategy(context.getProperty(KUSTO_AUTH_STRATEGY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppId(context.getProperty(APP_ID).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppKey(context.getProperty(APP_KEY).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setAppTenant(context.getProperty(APP_TENANT).evaluateAttributeExpressions().getValue());
+ adxConnectionParams.setKustoEngineURL(context.getProperty(CLUSTER_URL).evaluateAttributeExpressions().getValue());
+ if (this.executionClient != null) {
+ onStopped();
+ }
+
+ }
+
+ @OnStopped
+ public final void onStopped() {
+ if(this.executionClient!=null){
Review Comment:
```suggestion
if (this.executionClient != null) {
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx/src/test/java/org/apache/nifi/adx/TestAzureAdxSourceConnectionService.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.adx;
+
+import com.microsoft.azure.kusto.data.Client;
+import org.apache.nifi.adx.service.AzureAdxSourceConnectionService;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.ControllerServiceConfiguration;
+import org.apache.nifi.util.StandardProcessorTestRunner;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestAzureAdxSourceConnectionService {
+
+ private TestRunner runner;
+
+ private AzureAdxSourceConnectionService service;
+
+ private static final String MOCK_APP_ID = "mockAppId";
+
+ private static final String MOCK_APP_KEY = "mockAppKey";
+
+ private static final String MOCK_APP_TENANT = "mockAppTenant";
+
+ private static final String MOCK_CLUSTER_URL = "https://mockClusterUrl.com/";
+
+ @BeforeEach
+ public void setup() throws InitializationException {
+ runner = TestRunners.newTestRunner(new AbstractProcessor() {
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ List<PropertyDescriptor> propDescs = new ArrayList<>();
+ propDescs.add(new PropertyDescriptor.Builder()
+ .name("AdxService")
+ .description("AdxService")
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .required(true)
+ .build());
+ return propDescs;
+ }
+ });
Review Comment:
This can be simplified:
```suggestion
runner = TestRunners.newTestRunner(NoOpProcessor.class);
```
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/QueryAzureDataExplorer.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.adx.model.KustoQueryResponse;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParameter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = QueryAzureDataExplorer.ADX_QUERY_ERROR_MESSAGE, description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = QueryAzureDataExplorer.ADX_EXECUTED_QUERY, description = "Azure Data Explorer executed query.")
+})
+public class QueryAzureDataExplorer extends AbstractProcessor {
+ public static final String ADX_QUERY_ERROR_MESSAGE = "adx.query.error.message";
+ public static final String ADX_EXECUTED_QUERY = "adx.executed.query";
+
+ public static final Relationship SUCCESS = new Relationship.Builder()
+ .name("SUCCESS")
+ .description("Relationship for success")
+ .build();
+ public static final Relationship FAILED = new Relationship.Builder()
+ .name("FAILED")
+ .description("Relationship for failure")
+ .build();
+ public static final PropertyDescriptor DB_NAME = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.DB_NAME.name())
+ .displayName(AzureAdxSourceProcessorParameter.DB_NAME.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.DB_NAME.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_QUERY = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.ADX_QUERY.name())
+ .displayName(AzureAdxSourceProcessorParameter.ADX_QUERY.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.ADX_QUERY.getParamDescription())
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ADX_SOURCE_SERVICE = new PropertyDescriptor
+ .Builder().name(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.name())
+ .displayName(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.getParamDisplayName())
+ .description(AzureAdxSourceProcessorParameter.ADX_SOURCE_SERVICE.getParamDescription())
+ .required(true)
+ .identifiesControllerService(AdxSourceConnectionService.class)
+ .build();
+ private final ObjectMapper objectMapper = new ObjectMapper();
+ private final Set<Relationship> relationships = Set.of(SUCCESS,FAILED);
+ private final List<PropertyDescriptor> descriptors = List.of(ADX_SOURCE_SERVICE,DB_NAME,ADX_QUERY);
+ private AdxSourceConnectionService service;
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return this.relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ service = context.getProperty(ADX_SOURCE_SERVICE).asControllerService(AdxSourceConnectionService.class);
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile outgoingFlowFile;
+ String databaseName = context.getProperty(DB_NAME).getValue();
+ String adxQuery;
+ KustoQueryResponse kustoQueryResponse;
+
+ //checks if this processor has any preceding connection, if yes retrieve
+ if (context.hasIncomingConnection()) {
+ FlowFile incomingFlowFile = session.get();
+ //incoming connection exists but the incoming flowfile is null
+ if (incomingFlowFile == null && context.hasNonLoopConnection()) {
+ return;
+ }
+ //incoming connection exists and retrieve adxQuery from context
+ if (incomingFlowFile != null && incomingFlowFile.getSize() == 0) {
+ if (context.getProperty(ADX_QUERY).isSet()) {
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(incomingFlowFile).getValue();
+ } else {
+ String message = "FlowFile query is empty and no scheduled query is set";
+ getLogger().error(message);
+ incomingFlowFile = session.putAttribute(incomingFlowFile, ADX_QUERY_ERROR_MESSAGE, message);
+ session.transfer(incomingFlowFile, FAILED);
+ return;
+ }
+ } else {
+ try {
+ adxQuery = getQuery(session, incomingFlowFile);
+ } catch(IOException ioe) {
+ throw new ProcessException("Failed to read Query from FlowFile",ioe);
+ }
+ }
+ outgoingFlowFile = incomingFlowFile;
+ } else {
+ outgoingFlowFile = session.create();
+ adxQuery = context.getProperty(ADX_QUERY).evaluateAttributeExpressions(outgoingFlowFile).getValue();
+ }
+
+ try {
+ //execute Query
+ kustoQueryResponse = executeQuery(databaseName,adxQuery);
+ if(!kustoQueryResponse.isError()){
+ try(ByteArrayInputStream bais = new ByteArrayInputStream(objectMapper.writeValueAsBytes(kustoQueryResponse.getTableData()))){
+ session.importFrom(bais, outgoingFlowFile);
+ }
Review Comment:
This approach is very memory-intensive and needs to be changed. Serializing the entire result set into a byte array can easily lead to memory exhaustion for large numbers of results. The Kusto ClientFactory also includes a `StreamingClient` that returns an `InputStream`. Based on that option, both the Controller Service interface and this Processor implementation should be changed. The Service should return a streaming query result, which can be transferred directly to the FlowFile OutputStream using one of the methods on ProcessSession.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #7122:
URL: https://github.com/apache/nifi/pull/7122#discussion_r1157715107
##########
nifi-nar-bundles/nifi-adx-bundle/nifi-adx-processors/src/main/java/org/apache/nifi/processors/adx/AzureAdxSourceProcessor.java:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.adx;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.nifi.adx.AdxSourceConnectionService;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.adx.enums.AzureAdxSourceProcessorParamsEnum;
+import org.apache.nifi.processors.adx.enums.RelationshipStatusEnum;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+@Tags({"azure", "adx", "microsoft", "data", "explorer", "source"})
+@CapabilityDescription("This Processor acts as a ADX source connector which queries data from Azure Data Explorer."+
+ "This connector can act only as a start of the data pipeline getting data from ADX."+
+ "The queries which can be used further details can be found here https://learn.microsoft.com/en-us/azure/data-explorer/kusto/concepts/querylimits")
+@WritesAttributes({
+ @WritesAttribute(attribute = "ADX_QUERY_ERROR_MESSAGE", description = "Azure Data Explorer error message."),
+ @WritesAttribute(attribute = "ADX_EXECUTED_QUERY", description = "Azure Data Explorer executed query.")
+})
+public class AzureAdxSourceProcessor extends AbstractProcessor {
Review Comment:
Regarding the name, since `Adx` is an acronym that includes `Azure`, recommend renaming the Processor. Also, Processor implementation should not use the word `Processor`.
```suggestion
public class QueryAzureDataExplorer extends AbstractProcessor {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] kustonaut commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "kustonaut (via GitHub)" <gi...@apache.org>.
kustonaut commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1558556166
@exceptionfactory we have addressed the review comments. Request your re-review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tanmaya-panda1 commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "tanmaya-panda1 (via GitHub)" <gi...@apache.org>.
tanmaya-panda1 commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1600079607
Thanks a ton @exceptionfactory for reviewing the PR and implementing the fixes. I think we are good with all changes as suggested.
But I had a question though on QueryAzureDataExplorer. Since it mandatorily requires a flow file to query into ADX, can this processor become a starting point of a flow where we query into ADX using input query and write the output to a file or some other output processor?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory closed pull request #7122: NIFI-8294 Add Query Processor for Azure Data Explorer
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory closed pull request #7122: NIFI-8294 Add Query Processor for Azure Data Explorer
URL: https://github.com/apache/nifi/pull/7122
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] tanmaya-panda1 commented on pull request #7122: NIFI-8294 ADX as source for various NIFI destinations
Posted by "tanmaya-panda1 (via GitHub)" <gi...@apache.org>.
tanmaya-panda1 commented on PR #7122:
URL: https://github.com/apache/nifi/pull/7122#issuecomment-1590586984
sure, Thanks @exceptionfactory , please do let me know if anything is needed from my side.
--
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