Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-11992: Processor and sink service for filing tickets in Zendesk 

This closes #7644.

Signed-off-by: Peter Turcsanyi <turcsanyi@apache.org>
This commit is contained in:
Mark Bathori 2023-08-24 18:37:47 +02:00 committed by Peter Turcsanyi
parent 3b0d8100a3
commit 7770a17a6c
No known key found for this signature in database
GPG Key ID: 55A813F1C3E553DC
33 changed files with 2997 additions and 220 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
nifi-assembly/pom.xml
View File

@ -828,6 +828,12 @@ language governing permissions and limitations under the License. -->
<version>2.0.0-SNAPSHOT</version>
<type>nar</type>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-zendesk-services-nar</artifactId>
<version>2.0.0-SNAPSHOT</version>
<type>nar</type>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-dropbox-processors-nar</artifactId>

41
nifi-nar-bundles/nifi-extension-utils/nifi-record-path-property/pom.xml Normal file
View File

@ -0,0 +1,41 @@
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-extension-utils</artifactId>
<version>2.0.0-SNAPSHOT</version>
</parent>
<artifactId>nifi-record-path-property</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-record-path</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
</dependencies>
</project>

117
nifi-nar-bundles/nifi-extension-utils/nifi-record-path-property/src/main/java/org/apache/nifi/record/path/property/RecordPathPropertyUtil.java Normal file
View File

@ -0,0 +1,117 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.nifi.record.path.property;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.record.path.FieldValue;
import org.apache.nifi.record.path.RecordPath;
import org.apache.nifi.record.path.RecordPathResult;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.type.ChoiceDataType;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import static java.util.stream.Collectors.toList;
public final class RecordPathPropertyUtil {
private static final String NULL_VALUE = "null";
private static final Pattern RECORD_PATH_PATTERN = Pattern.compile("@\\{(/.*?)\\}");
private RecordPathPropertyUtil() {
}
/**
* Resolves the property value to be handled as a record path or a constant value.
*
* @param propertyValue property value to be resolved
* @param record record to resolve the value from if the property value is a record path
*/
public static String resolvePropertyValue(String propertyValue, Record record) {
if (propertyValue != null && !propertyValue.isBlank()) {
final Matcher matcher = RECORD_PATH_PATTERN.matcher(propertyValue);
if (matcher.matches()) {
return resolveRecordState(matcher.group(1), record);
} else {
return propertyValue;
}
}
return null;
}
private static String resolveRecordState(String pathValue, final Record record) {
final RecordPath recordPath = RecordPath.compile(pathValue);
final RecordPathResult result = recordPath.evaluate(record);
final List<FieldValue> fieldValues = result.getSelectedFields().collect(toList());
final FieldValue fieldValue = getMatchingFieldValue(recordPath, fieldValues);
if (fieldValue.getValue() == null || fieldValue.getValue() == NULL_VALUE) {
return null;
}
return getFieldValue(recordPath, fieldValue);
}
/**
* The method checks if only one result were received for the give record path.
*
* @param recordPath path to the requested field
* @param resultList result list
* @return matching field
*/
private static FieldValue getMatchingFieldValue(final RecordPath recordPath, final List<FieldValue> resultList) {
if (resultList.isEmpty()) {
throw new ProcessException(String.format("Evaluated RecordPath [%s] against Record but got no results", recordPath));
}
if (resultList.size() > 1) {
throw new ProcessException(String.format("Evaluated RecordPath [%s] against Record and received multiple distinct results [%s]", recordPath, resultList));
}
return resultList.get(0);
}
/**
* The method checks the field's type and filters out every non-compatible type.
*
* @param recordPath path to the requested field
* @param fieldValue record field
* @return value of the record field
*/
private static String getFieldValue(final RecordPath recordPath, FieldValue fieldValue) {
final RecordFieldType fieldType = fieldValue.getField().getDataType().getFieldType();
if (fieldType == RecordFieldType.RECORD || fieldType == RecordFieldType.ARRAY || fieldType == RecordFieldType.MAP) {
throw new ProcessException(String.format("The provided RecordPath [%s] points to a [%s] type value", recordPath, fieldType));
}
if (fieldType == RecordFieldType.CHOICE) {
final ChoiceDataType choiceDataType = (ChoiceDataType) fieldValue.getField().getDataType();
final List<DataType> possibleTypes = choiceDataType.getPossibleSubTypes();
if (possibleTypes.stream().anyMatch(type -> type.getFieldType() == RecordFieldType.RECORD)) {
throw new ProcessException(String.format("The provided RecordPath [%s] points to a [CHOICE] type value with Record subtype", recordPath));
}
}
return String.valueOf(fieldValue.getValue());
}
}

1
nifi-nar-bundles/nifi-extension-utils/pom.xml
View File

@ -45,5 +45,6 @@
<module>nifi-service-utils</module>
<module>nifi-syslog-utils</module>
<module>nifi-conflict-resolution</module>
<module>nifi-record-path-property</module>
</modules>
</project>

65
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/pom.xml Normal file
View File

@ -0,0 +1,65 @@
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-zendesk-bundle</artifactId>
<version>2.0.0-SNAPSHOT</version>
</parent>
<artifactId>nifi-zendesk-common</artifactId>
<packaging>jar</packaging>
<dependencies>
<!-- Internal dependencies -->
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-utils</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-web-client-provider-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-record-path-property</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
<!-- External dependencies -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<!-- Test dependencies -->
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-ssl-context-service-api</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

48
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/src/main/java/org/apache/nifi/common/zendesk/ZendeskAuthenticationContext.java Normal file
View File

@ -0,0 +1,48 @@
/*
* 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.common.zendesk;
public class ZendeskAuthenticationContext {
private final String subdomain;
private final String user;
private final ZendeskAuthenticationType authenticationType;
private final String authenticationCredentials;
public ZendeskAuthenticationContext(String subdomain, String user, ZendeskAuthenticationType authenticationType, String authenticationCredentials) {
this.subdomain = subdomain;
this.user = user;
this.authenticationType = authenticationType;
this.authenticationCredentials =authenticationCredentials;
}
public String getSubdomain() {
return subdomain;
}
public String getUser() {
return user;
}
public ZendeskAuthenticationType getAuthenticationType() {
return authenticationType;
}
public String getAuthenticationCredentials() {
return authenticationCredentials;
}
}

7
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/main/java/org/apache/nifi/processors/zendesk/ZendeskAuthenticationType.java → nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/src/main/java/org/apache/nifi/common/zendesk/ZendeskAuthenticationType.java
View File

@ -15,12 +15,13 @@
* limitations under the License.
*/
package org.apache.nifi.processors.zendesk;
package org.apache.nifi.common.zendesk;
import static java.lang.String.format;
import org.apache.nifi.components.DescribedValue;
import java.util.stream.Stream;
import org.apache.nifi.components.DescribedValue;
import static java.lang.String.format;
public enum ZendeskAuthenticationType implements DescribedValue {
PASSWORD("password", "Password",

102
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/src/main/java/org/apache/nifi/common/zendesk/ZendeskClient.java Normal file
View File

@ -0,0 +1,102 @@
/*
* 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.common.zendesk;
import org.apache.nifi.web.client.api.HttpResponseEntity;
import org.apache.nifi.web.client.api.HttpUriBuilder;
import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.OptionalLong;
import static java.lang.String.format;
import static java.util.Base64.getEncoder;
import static org.apache.nifi.common.zendesk.ZendeskProperties.HTTPS;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_HOST_TEMPLATE;
public class ZendeskClient {
private static final String AUTHORIZATION_HEADER_NAME = "Authorization";
private static final String CONTENT_TYPE_HEADER_NAME = "Content-Type";
private static final String BASIC_AUTH_PREFIX = "Basic ";
private final WebClientServiceProvider webClientServiceProvider;
private final ZendeskAuthenticationContext authenticationContext;
public ZendeskClient(WebClientServiceProvider webClientServiceProvider, ZendeskAuthenticationContext authenticationContext) {
this.webClientServiceProvider = webClientServiceProvider;
this.authenticationContext = authenticationContext;
}
/**
* Sends a POST request to the Zendesk API.
*
* @param uri target uri
* @param inputStream body of the request
* @return response from the Zendesk API
* @throws IOException error while performing the POST request
*/
public HttpResponseEntity performPostRequest(URI uri, InputStream inputStream) throws IOException {
return webClientServiceProvider.getWebClientService()
.post()
.uri(uri)
.header(CONTENT_TYPE_HEADER_NAME, ZendeskProperties.APPLICATION_JSON)
.header(AUTHORIZATION_HEADER_NAME, basicAuthHeaderValue())
.body(inputStream, OptionalLong.of(inputStream.available()))
.retrieve();
}
/**
* Sends a GET request to the Zendesk API.
*
* @param uri target uri
* @return response from the Zendesk API
*/
public HttpResponseEntity performGetRequest(URI uri) {
return webClientServiceProvider.getWebClientService()
.get()
.uri(uri)
.header(AUTHORIZATION_HEADER_NAME, basicAuthHeaderValue())
.retrieve();
}
/**
* Creates a Uri builder with the provided resource path.
*
* @param resourcePath resource path
* @return Uri builder
*/
public HttpUriBuilder uriBuilder(String resourcePath) {
return webClientServiceProvider.getHttpUriBuilder()
.scheme(HTTPS)
.host(format(ZENDESK_HOST_TEMPLATE, authenticationContext.getSubdomain()))
.encodedPath(resourcePath);
}
/**
* Constructs the header for the Zendesk API call.
*
* @return the constructed header
*/
private String basicAuthHeaderValue() {
final String user = authenticationContext.getAuthenticationType().enrichUserName(authenticationContext.getUser());
final String userWithPassword = user + ":" + authenticationContext.getAuthenticationCredentials();
return BASIC_AUTH_PREFIX + getEncoder().encodeToString(userWithPassword.getBytes());
}
}

126
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/src/main/java/org/apache/nifi/common/zendesk/ZendeskProperties.java Normal file
View File

@ -0,0 +1,126 @@
/*
* 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.common.zendesk;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
import static org.apache.nifi.expression.ExpressionLanguageScope.FLOWFILE_ATTRIBUTES;
public final class ZendeskProperties {
public static final String WEB_CLIENT_SERVICE_PROVIDER_NAME = "web-client-service-provider";
public static final String ZENDESK_SUBDOMAIN_NAME = "zendesk-subdomain";
public static final String ZENDESK_USER_NAME = "zendesk-user";
public static final String ZENDESK_AUTHENTICATION_TYPE_NAME = "zendesk-authentication-type-name";
public static final String ZENDESK_AUTHENTICATION_CREDENTIAL_NAME = "zendesk-authentication-value-name";
public static final String ZENDESK_TICKET_COMMENT_BODY_NAME = "zendesk-comment-body";
public static final String ZENDESK_TICKET_SUBJECT_NAME = "zendesk-subject";
public static final String ZENDESK_TICKET_PRIORITY_NAME = "zendesk-priority";
public static final String ZENDESK_TICKET_TYPE_NAME = "zendesk-type";
public static final String HTTPS = "https";
public static final String APPLICATION_JSON = "application/json";
public static final String ZENDESK_HOST_TEMPLATE = "%s.zendesk.com";
public static final String ZENDESK_CREATE_TICKET_RESOURCE = "/api/v2/tickets";
public static final String ZENDESK_CREATE_TICKETS_RESOURCE = "/api/v2/tickets/create_many";
public static final String ZENDESK_TICKET_ROOT_NODE = "/ticket";
public static final String ZENDESK_TICKETS_ROOT_NODE = "/tickets";
public static final String REL_SUCCESS_NAME = "success";
public static final String REL_FAILURE_NAME = "failure";
private ZendeskProperties() {}
public static final PropertyDescriptor WEB_CLIENT_SERVICE_PROVIDER = new PropertyDescriptor.Builder()
.name(WEB_CLIENT_SERVICE_PROVIDER_NAME)
.displayName("Web Client Service Provider")
.description("Controller service for HTTP client operations.")
.identifiesControllerService(WebClientServiceProvider.class)
.required(true)
.build();
public static final PropertyDescriptor ZENDESK_SUBDOMAIN = new PropertyDescriptor.Builder()
.name(ZENDESK_SUBDOMAIN_NAME)
.displayName("Subdomain Name")
.description("Name of the Zendesk subdomain.")
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.required(true)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.build();
public static final PropertyDescriptor ZENDESK_USER = new PropertyDescriptor.Builder()
.name(ZENDESK_USER_NAME)
.displayName("User Name")
.description("Login user to Zendesk subdomain.")
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.required(true)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.build();
public static final PropertyDescriptor ZENDESK_AUTHENTICATION_TYPE = new PropertyDescriptor.Builder()
.name(ZENDESK_AUTHENTICATION_TYPE_NAME)
.displayName("Authentication Type")
.description("Type of authentication to Zendesk API.")
.required(true)
.allowableValues(ZendeskAuthenticationType.class)
.build();
public static final PropertyDescriptor ZENDESK_AUTHENTICATION_CREDENTIAL = new PropertyDescriptor.Builder()
.name(ZENDESK_AUTHENTICATION_CREDENTIAL_NAME)
.displayName("Authentication Credential")
.description("Password or authentication token for Zendesk login user.")
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.sensitive(true)
.required(true)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();
public static final PropertyDescriptor ZENDESK_TICKET_COMMENT_BODY = new PropertyDescriptor.Builder()
.name(ZENDESK_TICKET_COMMENT_BODY_NAME)
.displayName("Comment Body")
.description("The content or the path to the comment body in the incoming record.")
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.required(true)
.build();
public static final PropertyDescriptor ZENDESK_TICKET_SUBJECT = new PropertyDescriptor.Builder()
.name(ZENDESK_TICKET_SUBJECT_NAME)
.displayName("Subject")
.description("The content or the path to the subject in the incoming record.")
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.build();
public static final PropertyDescriptor ZENDESK_TICKET_PRIORITY = new PropertyDescriptor.Builder()
.name(ZENDESK_TICKET_PRIORITY_NAME)
.displayName("Priority")
.description("The content or the path to the priority in the incoming record.")
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.build();
public static final PropertyDescriptor ZENDESK_TICKET_TYPE = new PropertyDescriptor.Builder()
.name(ZENDESK_TICKET_TYPE_NAME)
.displayName("Type")
.description("The content or the path to the type in the incoming record.")
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.build();
}

113
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/src/main/java/org/apache/nifi/common/zendesk/util/ZendeskRecordPathUtils.java Normal file
View File

@ -0,0 +1,113 @@
/*
* 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.common.zendesk.util;
import com.fasterxml.jackson.core.JsonPointer;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.TextNode;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.record.path.property.RecordPathPropertyUtil;
import org.apache.nifi.serialization.record.Record;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKETS_ROOT_NODE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_ROOT_NODE;
public final class ZendeskRecordPathUtils {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private ZendeskRecordPathUtils() {
}
/**
* Resolves the property value and appends it as a new node at the given path.
*
* @param path path in the request object
* @param value property value to be resolved
* @param baseTicketNode base request object where the field value will be added
* @param record record to receive the value from if the field value is a record path
*/
public static void addField(String path, String value, ObjectNode baseTicketNode, Record record) {
final String resolvedValue = RecordPathPropertyUtil.resolvePropertyValue(value, record);
if (resolvedValue != null) {
addNewNodeAtPath(baseTicketNode, JsonPointer.compile(path), new TextNode(resolvedValue));
}
}
/**
* Adds a user defined dynamic field to the request object. If the user specifies the path in the request object as full path (starting with '/ticket' or '/tickets')
* the method removes them since the root path is specified later based on the number of records.
*
* @param path path in the request object
* @param value dynamic field value
* @param baseTicketNode base request object where the field value will be added
* @param record record to receive the value from if the field value is a record path
*/
public static void addDynamicField(String path, String value, ObjectNode baseTicketNode, Record record) {
if (path.startsWith(ZENDESK_TICKET_ROOT_NODE)) {
path = path.substring(7);
} else if (path.startsWith(ZENDESK_TICKETS_ROOT_NODE)) {
path = path.substring(8);
}
addField(path, value, baseTicketNode, record);
}
/**
* Adds a new node on the provided path with the give value to the request object.
*
* @param baseNode base object where the new node will be added
* @param path path of the new node
* @param value value of the new node
*/
public static void addNewNodeAtPath(final ObjectNode baseNode, final JsonPointer path, final JsonNode value) {
final JsonPointer parentPointer = path.head();
final String fieldName = path.last().toString().substring(1);
JsonNode parentNode = getOrCreateParentNode(baseNode, parentPointer, fieldName);
setNodeValue(value, fieldName, parentNode);
}
private static void setNodeValue(JsonNode value, String fieldName, JsonNode parentNode) {
if (parentNode.isArray()) {
ArrayNode arrayNode = (ArrayNode) parentNode;
int index = Integer.parseInt(fieldName);
for (int i = arrayNode.size(); i <= index; i++) {
arrayNode.addNull();
}
arrayNode.set(index, value);
} else if (parentNode.isObject()) {
((ObjectNode) parentNode).set(fieldName, value);
} else {
throw new IllegalArgumentException("Unsupported node type" + parentNode.getNodeType().name());
}
}
private static JsonNode getOrCreateParentNode(ObjectNode rootNode, JsonPointer parentPointer, String fieldName) {
JsonNode parentNode = rootNode.at(parentPointer);
if (parentNode.isMissingNode() || parentNode.isNull()) {
parentNode = StringUtils.isNumeric(fieldName) ? OBJECT_MAPPER.createArrayNode() : OBJECT_MAPPER.createObjectNode();
addNewNodeAtPath(rootNode, parentPointer, parentNode);
}
return parentNode;
}
}

104
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/src/main/java/org/apache/nifi/common/zendesk/util/ZendeskUtils.java Normal file
View File

@ -0,0 +1,104 @@
/*
* 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.common.zendesk.util;
import com.fasterxml.jackson.core.JsonPointer;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.common.zendesk.ZendeskProperties;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.web.client.api.HttpResponseEntity;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
public final class ZendeskUtils {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private ZendeskUtils() {
}
/**
* Gets the body from the Http response.
*
* @param response Http response
* @return response body
*/
public static String getResponseBody(HttpResponseEntity response) {
try (InputStream responseBodyStream = response.body()) {
return new String(responseBodyStream.readAllBytes(), StandardCharsets.UTF_8);
} catch (IOException e) {
throw new UncheckedIOException("Reading response body has failed", e);
}
}
/**
* Collects every non-blank dynamic property from the context.
*
* @param context property context
* @param properties property list from context
* @param attributes attributes that Expressions can reference
* @return Map of dynamic properties
*/
public static Map<String, String> getDynamicProperties(PropertyContext context, Map<PropertyDescriptor, String> properties, Map<String, String> attributes) {
return properties.entrySet().stream()
// filter non-blank dynamic properties
.filter(e -> e.getKey().isDynamic()
&& StringUtils.isNotBlank(e.getValue())
&& StringUtils.isNotBlank(context.getProperty(e.getKey()).evaluateAttributeExpressions(attributes).getValue())
)
// convert to Map keys and evaluated property values
.collect(Collectors.toMap(
e -> e.getKey().getName(),
e -> context.getProperty(e.getKey()).evaluateAttributeExpressions(attributes).getValue()
));
}
/**
* Creates the request object for the Zendesk ticket creation. The request objects root node will be created based on the number of tickets created.
*
* @param zendeskTickets list of tickets to be sent to the Zendesk API
* @return input stream of the request object
* @throws JsonProcessingException error while processing the request object
*/
public static InputStream createRequestObject(List<ObjectNode> zendeskTickets) throws JsonProcessingException {
ObjectNode ticketNode = OBJECT_MAPPER.createObjectNode();
if (zendeskTickets.size() > 1) {
ArrayNode arrayNode = OBJECT_MAPPER.createArrayNode();
for (ObjectNode ticket : zendeskTickets) {
arrayNode.add(ticket);
}
ZendeskRecordPathUtils.addNewNodeAtPath(ticketNode, JsonPointer.compile(ZendeskProperties.ZENDESK_TICKETS_ROOT_NODE), arrayNode);
} else {
ZendeskRecordPathUtils.addNewNodeAtPath(ticketNode, JsonPointer.compile(ZendeskProperties.ZENDESK_TICKET_ROOT_NODE), zendeskTickets.get(0));
}
return new ByteArrayInputStream(OBJECT_MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(ticketNode).getBytes());
}
}

46
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/src/main/java/org/apache/nifi/common/zendesk/validation/JsonPointerPropertyNameValidator.java Normal file
View File

@ -0,0 +1,46 @@
/*
* 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.common.zendesk.validation;
import com.fasterxml.jackson.core.JsonPointer;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.Validator;
import org.apache.nifi.record.path.exception.RecordPathException;
public class JsonPointerPropertyNameValidator implements Validator {
@Override
public ValidationResult validate(final String subject, final String input, final ValidationContext context) {
try {
JsonPointer.compile(subject);
return new ValidationResult.Builder()
.input(input)
.subject(subject)
.valid(true)
.explanation("Valid JsonPointer")
.build();
} catch (final RecordPathException e) {
return new ValidationResult.Builder()
.input(input)
.subject(subject)
.valid(false)
.explanation("Property Name is not a valid JsonPointer value: " + e.getMessage())
.build();
}
}
}

102
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-common/src/test/java/org/apache/nifi/common/zendesk/ZendeskRecordPathUtilsTest.java Normal file
View File

@ -0,0 +1,102 @@
/*
* 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.common.zendesk;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.type.ArrayDataType;
import org.apache.nifi.serialization.record.type.RecordDataType;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.apache.nifi.common.zendesk.util.ZendeskRecordPathUtils.addField;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class ZendeskRecordPathUtilsTest {
private final ObjectMapper mapper = new ObjectMapper();
@Test
public void testFieldValueEvaluation() {
ObjectNode testNode;
Record record = initRecord();
testNode = mapper.createObjectNode();
addField("/a/b", "@{/field1}", testNode, record);
Assertions.assertEquals("{\"a\":{\"b\":\"value1\"}}", testNode.toString());
testNode = mapper.createObjectNode();
addField("/a/b/c", "constant", testNode, record);
Assertions.assertEquals("{\"a\":{\"b\":{\"c\":\"constant\"}}}", testNode.toString());
testNode = mapper.createObjectNode();
addField("/a/0", "array_element", testNode, record);
Assertions.assertEquals("{\"a\":[\"array_element\"]}", testNode.toString());
ProcessException e1 = assertThrows(ProcessException.class, () -> addField("/a", "@{/field2}", mapper.createObjectNode(), record));
Assertions.assertEquals("The provided RecordPath [/field2] points to a [ARRAY] type value", e1.getMessage());
ProcessException e2 = assertThrows(ProcessException.class, () -> addField("/a", "@{/field3}", mapper.createObjectNode(), record));
Assertions.assertEquals("The provided RecordPath [/field3] points to a [RECORD] type value", e2.getMessage());
ProcessException e3 = assertThrows(ProcessException.class, () -> addField("/a", "@{/field4}", mapper.createObjectNode(), record));
Assertions.assertEquals("The provided RecordPath [/field4] points to a [CHOICE] type value with Record subtype", e3.getMessage());
}
private Record initRecord() {
List<RecordField> recordFields = new ArrayList<>();
recordFields.add(new RecordField("nestedField1", RecordFieldType.STRING.getDataType()));
recordFields.add(new RecordField("nestedField2", RecordFieldType.STRING.getDataType()));
RecordSchema nestedRecordSchema = new SimpleRecordSchema(recordFields);
List<RecordField> fields = new ArrayList<>();
fields.add(new RecordField("field1", RecordFieldType.STRING.getDataType()));
fields.add(new RecordField("field2", new ArrayDataType(RecordFieldType.STRING.getDataType())));
fields.add(new RecordField("field3", new RecordDataType(nestedRecordSchema)));
fields.add(new RecordField("field4", RecordFieldType.CHOICE.getChoiceDataType(
RecordFieldType.STRING.getDataType(), RecordFieldType.INT.getDataType(), RecordFieldType.RECORD.getDataType())));
RecordSchema schema = new SimpleRecordSchema(fields);
List<String> valueList = new ArrayList<>();
valueList.add("listElement");
Map<String, Object> nestedValueMap = new HashMap<>();
nestedValueMap.put("nestedField1", "nestedValue1");
nestedValueMap.put("nestedField2", "nestedValue2");
Map<String, Object> valueMap = new HashMap<>();
valueMap.put("field1", "value1");
valueMap.put("field2", valueList);
valueMap.put("field3", nestedValueMap);
valueMap.put("field4", "field4");
return new MapRecord(schema, valueMap);
}
}

34
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-nar/src/main/resources/META-INF/LICENSE
View File

@ -201,4 +201,38 @@
See the License for the specific language governing permissions and
limitations under the License.
APACHE NIFI SUBCOMPONENTS:
The Apache NiFi project contains subcomponents with separate copyright
notices and license terms. Your use of the source code for the these
subcomponents is subject to the terms and conditions of the following
licenses.
The binary distribution of this product bundles 'Antlr 3' which is available
under a "3-clause BSD" license. For details see http://www.antlr3.org/license.html
Copyright (c) 2010 Terence Parr
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
Neither the name of the author nor the names of its contributors may be used
to endorse or promote products derived from this software without specific
prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
THE POSSIBILITY OF SUCH DAMAGE.

36
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-nar/src/main/resources/META-INF/NOTICE
View File

@ -2,38 +2,4 @@ nifi-zendesk-nar
Copyright 2015-2022 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
===========================================
Apache Software License v2
===========================================
The following binary components are provided under the Apache Software License v2
(ASLv2) Apache Commons IO
The following NOTICE information applies:
Apache Commons IO
Copyright 2002-2017 The Apache Software Foundation
(ASLv2) Jackson JSON processor
The following NOTICE information applies:
# Jackson JSON processor
Jackson is a high-performance, Free/Open Source JSON processing library.
It was originally written by Tatu Saloranta (tatu.saloranta@iki.fi), and has
been in development since 2007.
It is currently developed by a community of developers, as well as supported
commercially by FasterXML.com.
## Licensing
Jackson core and extension components may licensed under different licenses.
To find the details that apply to this artifact see the accompanying LICENSE file.
For more information, including possible other licensing options, contact
FasterXML.com (http://fasterxml.com).
## Credits
A list of contributors may be found from CREDITS file, which is included
in some artifacts (usually source distributions); but is always available
from the source code management (SCM) system project uses.
The Apache Software Foundation (http://www.apache.org/).

24
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/pom.xml
View File

@ -13,7 +13,8 @@
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">
<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>
@ -26,13 +27,11 @@
<packaging>jar</packaging>
<dependencies>
<!-- Internal dependencies -->
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-utils</artifactId>
<artifactId>nifi-zendesk-common</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
@ -41,12 +40,8 @@
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-record-serialization-service-api</artifactId>
</dependency>
<!-- Test dependencies -->
@ -75,5 +70,10 @@
<artifactId>mockwebserver</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-mock-record-utils</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

62
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/main/java/org/apache/nifi/processors/zendesk/AbstractZendesk.java Normal file
View File

@ -0,0 +1,62 @@
/*
* 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.zendesk;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.common.zendesk.ZendeskAuthenticationContext;
import org.apache.nifi.common.zendesk.ZendeskAuthenticationType;
import org.apache.nifi.common.zendesk.ZendeskClient;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.web.client.api.HttpUriBuilder;
import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
import static org.apache.nifi.common.zendesk.ZendeskProperties.REL_SUCCESS_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.WEB_CLIENT_SERVICE_PROVIDER;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_CREDENTIAL;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_TYPE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_SUBDOMAIN;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_USER;
public abstract class AbstractZendesk extends AbstractProcessor {
static final String RECORD_COUNT_ATTRIBUTE_NAME = "record.count";
ZendeskClient zendeskClient;
static final Relationship REL_SUCCESS = new Relationship.Builder()
.name(REL_SUCCESS_NAME)
.description("For FlowFiles created as a result of a successful HTTP request.")
.build();
@OnScheduled
public void onScheduled(ProcessContext context) {
final WebClientServiceProvider webClientServiceProvider = context.getProperty(WEB_CLIENT_SERVICE_PROVIDER).asControllerService(WebClientServiceProvider.class);
final String user = context.getProperty(ZENDESK_USER).evaluateAttributeExpressions().getValue();
final ZendeskAuthenticationType authenticationType = ZendeskAuthenticationType.forName(context.getProperty(ZENDESK_AUTHENTICATION_TYPE).getValue());
final String authenticationCredentials = context.getProperty(ZENDESK_AUTHENTICATION_CREDENTIAL).evaluateAttributeExpressions().getValue();
final String subdomain = context.getProperty(ZENDESK_SUBDOMAIN).evaluateAttributeExpressions().getValue();
final ZendeskAuthenticationContext authenticationContext = new ZendeskAuthenticationContext(subdomain, user, authenticationType, authenticationCredentials);
zendeskClient = new ZendeskClient(webClientServiceProvider, authenticationContext);
}
HttpUriBuilder uriBuilder(String resourcePath) {
return zendeskClient.uriBuilder(resourcePath);
}
}

179
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/main/java/org/apache/nifi/processors/zendesk/GetZendesk.java
View File

@ -17,42 +17,12 @@
package org.apache.nifi.processors.zendesk;
import static com.fasterxml.jackson.core.JsonEncoding.UTF8;
import static com.fasterxml.jackson.core.JsonToken.FIELD_NAME;
import static com.fasterxml.jackson.core.JsonToken.VALUE_NULL;
import static java.lang.String.format;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Base64.getEncoder;
import static java.util.Collections.singleton;
import static java.util.Collections.singletonMap;
import static java.util.stream.Collectors.collectingAndThen;
import static java.util.stream.Collectors.toList;
import static org.apache.nifi.annotation.behavior.InputRequirement.Requirement.INPUT_FORBIDDEN;
import static org.apache.nifi.components.state.Scope.CLUSTER;
import static org.apache.nifi.expression.ExpressionLanguageScope.FLOWFILE_ATTRIBUTES;
import static org.apache.nifi.processor.util.StandardValidators.NON_BLANK_VALIDATOR;
import static org.apache.nifi.processor.util.StandardValidators.NON_EMPTY_VALIDATOR;
import static org.apache.nifi.processor.util.StandardValidators.POSITIVE_LONG_VALIDATOR;
import static org.apache.nifi.processors.zendesk.GetZendesk.RECORD_COUNT_ATTRIBUTE_NAME;
import static org.apache.nifi.web.client.api.HttpResponseStatus.OK;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
import org.apache.commons.io.IOUtils;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.PrimaryNodeOnly;
import org.apache.nifi.annotation.behavior.Stateful;
@ -63,12 +33,10 @@ import org.apache.nifi.annotation.configuration.DefaultSchedule;
import org.apache.nifi.annotation.configuration.DefaultSettings;
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.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
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;
@ -77,7 +45,36 @@ import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.web.client.api.HttpResponseEntity;
import org.apache.nifi.web.client.api.HttpUriBuilder;
import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
import static com.fasterxml.jackson.core.JsonEncoding.UTF8;
import static com.fasterxml.jackson.core.JsonToken.FIELD_NAME;
import static com.fasterxml.jackson.core.JsonToken.VALUE_NULL;
import static java.util.Collections.singleton;
import static java.util.Collections.singletonMap;
import static java.util.stream.Collectors.collectingAndThen;
import static java.util.stream.Collectors.toList;
import static org.apache.nifi.annotation.behavior.InputRequirement.Requirement.INPUT_FORBIDDEN;
import static org.apache.nifi.common.zendesk.ZendeskProperties.WEB_CLIENT_SERVICE_PROVIDER;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_CREDENTIAL;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_TYPE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_SUBDOMAIN;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_USER;
import static org.apache.nifi.common.zendesk.util.ZendeskUtils.getResponseBody;
import static org.apache.nifi.components.state.Scope.CLUSTER;
import static org.apache.nifi.expression.ExpressionLanguageScope.FLOWFILE_ATTRIBUTES;
import static org.apache.nifi.processor.util.StandardValidators.POSITIVE_LONG_VALIDATOR;
import static org.apache.nifi.processors.zendesk.GetZendesk.RECORD_COUNT_ATTRIBUTE_NAME;
import static org.apache.nifi.web.client.api.HttpResponseStatus.OK;
@PrimaryNodeOnly
@TriggerSerially
@ -89,80 +86,19 @@ import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
@WritesAttributes({
@WritesAttribute(attribute = RECORD_COUNT_ATTRIBUTE_NAME, description = "The number of records fetched by the processor.")})
@DefaultSchedule(strategy = SchedulingStrategy.TIMER_DRIVEN, period = "1 min")
public class GetZendesk extends AbstractProcessor {
public class GetZendesk extends AbstractZendesk {
static final int HTTP_TOO_MANY_REQUESTS = 429;
static final String RECORD_COUNT_ATTRIBUTE_NAME = "record.count";
static final String REL_SUCCESS_NAME = "success";
static final String WEB_CLIENT_SERVICE_PROVIDER_NAME = "web-client-service-provider";
static final String ZENDESK_SUBDOMAIN_NAME = "zendesk-subdomain";
static final String ZENDESK_USER_NAME = "zendesk-user";
static final String ZENDESK_AUTHENTICATION_TYPE_NAME = "zendesk-authentication-type-name";
static final String ZENDESK_AUTHENTICATION_CREDENTIAL_NAME = "zendesk-authentication-value-name";
static final String ZENDESK_EXPORT_METHOD_NAME = "zendesk-export-method";
static final String ZENDESK_RESOURCE_NAME = "zendesk-resource";
static final String ZENDESK_QUERY_START_TIMESTAMP_NAME = "zendesk-query-start-timestamp";
private static final String HTTPS = "https";
private static final String AUTHORIZATION_HEADER_NAME = "Authorization";
private static final String BASIC_AUTH_PREFIX = "Basic ";
private static final String ZENDESK_HOST_TEMPLATE = "%s.zendesk.com";
private static final Relationship REL_SUCCESS = new Relationship.Builder()
.name(REL_SUCCESS_NAME)
.description("For FlowFiles created as a result of a successful HTTP request.")
.build();
private static final Set<Relationship> RELATIONSHIPS = singleton(REL_SUCCESS);
private static final PropertyDescriptor WEB_CLIENT_SERVICE_PROVIDER = new PropertyDescriptor.Builder()
.name(WEB_CLIENT_SERVICE_PROVIDER_NAME)
.displayName("Web Client Service Provider")
.description("Controller service for HTTP client operations.")
.identifiesControllerService(WebClientServiceProvider.class)
.required(true)
.build();
private static final PropertyDescriptor ZENDESK_SUBDOMAIN = new PropertyDescriptor.Builder()
.name(ZENDESK_SUBDOMAIN_NAME)
.displayName("Zendesk Subdomain Name")
.description("Name of the Zendesk subdomain.")
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.required(true)
.addValidator(NON_BLANK_VALIDATOR)
.build();
private static final PropertyDescriptor ZENDESK_USER = new PropertyDescriptor.Builder()
.name(ZENDESK_USER_NAME)
.displayName("Zendesk User Name")
.description("Login user to Zendesk subdomain.")
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.required(true)
.addValidator(NON_BLANK_VALIDATOR)
.build();
private static final PropertyDescriptor ZENDESK_AUTHENTICATION_TYPE = new PropertyDescriptor.Builder()
.name(ZENDESK_AUTHENTICATION_TYPE_NAME)
.displayName("Zendesk Authentication Type")
.description("Type of authentication to Zendesk API.")
.required(true)
.allowableValues(ZendeskAuthenticationType.class)
.build();
private static final PropertyDescriptor ZENDESK_AUTHENTICATION_CREDENTIAL = new PropertyDescriptor.Builder()
.name(ZENDESK_AUTHENTICATION_CREDENTIAL_NAME)
.displayName("Zendesk Authentication Credential")
.description("Password or authentication token for Zendesk login user.")
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
.sensitive(true)
.required(true)
.addValidator(NON_EMPTY_VALIDATOR)
.build();
private static final PropertyDescriptor ZENDESK_EXPORT_METHOD = new PropertyDescriptor.Builder()
.name(ZENDESK_EXPORT_METHOD_NAME)
.displayName("Zendesk Export Method")
.displayName("Export Method")
.description("Method for incremental export.")
.required(true)
.allowableValues(ZendeskExportMethod.class)
@ -170,7 +106,7 @@ public class GetZendesk extends AbstractProcessor {
private static final PropertyDescriptor ZENDESK_RESOURCE = new PropertyDescriptor.Builder()
.name(ZENDESK_RESOURCE_NAME)
.displayName("Zendesk Resource")
.displayName("Resource")
.description("The particular Zendesk resource which is meant to be exported.")
.required(true)
.allowableValues(ZendeskResource.class)
@ -178,7 +114,7 @@ public class GetZendesk extends AbstractProcessor {
private static final PropertyDescriptor ZENDESK_QUERY_START_TIMESTAMP = new PropertyDescriptor.Builder()
.name(ZENDESK_QUERY_START_TIMESTAMP_NAME)
.displayName("Zendesk Query Start Timestamp")
.displayName("Query Start Timestamp")
.description("Initial timestamp to query Zendesk API from in Unix timestamp seconds format.")
.addValidator(POSITIVE_LONG_VALIDATOR)
.expressionLanguageSupported(FLOWFILE_ATTRIBUTES)
@ -199,8 +135,6 @@ public class GetZendesk extends AbstractProcessor {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static final JsonFactory JSON_FACTORY = OBJECT_MAPPER.getFactory();
private volatile WebClientServiceProvider webClientServiceProvider;
@Override
public Set<Relationship> getRelationships() {
return RELATIONSHIPS;
@ -228,18 +162,13 @@ public class GetZendesk extends AbstractProcessor {
return results;
}
@OnScheduled
public void onScheduled(ProcessContext context) {
webClientServiceProvider = context.getProperty(WEB_CLIENT_SERVICE_PROVIDER).asControllerService(WebClientServiceProvider.class);
}
@Override
public void onTrigger(ProcessContext context, ProcessSession session) {
ZendeskResource zendeskResource = ZendeskResource.forName(context.getProperty(ZENDESK_RESOURCE).getValue());
ZendeskExportMethod exportMethod = ZendeskExportMethod.forName(context.getProperty(ZENDESK_EXPORT_METHOD).getValue());
URI uri = createUri(context, zendeskResource, exportMethod);
HttpResponseEntity response = performQuery(context, uri);
HttpResponseEntity response = zendeskClient.performGetRequest(uri);
if (response.statusCode() == OK.getCode()) {
AtomicInteger resultCount = new AtomicInteger(0);
@ -258,15 +187,14 @@ public class GetZendesk extends AbstractProcessor {
getLogger().error("Rate limit exceeded for uri={}, yielding before retrying request.", uri);
context.yield();
} else {
getLogger().error("HTTP {} error for uri={} with response={}, yielding before retrying request.", response.statusCode(), uri, responseBodyToString(context, response));
getLogger().error("HTTP {} error for uri={} with response={}, yielding before retrying request.", response.statusCode(), uri, getResponseBody(response));
context.yield();
}
}
private URI createUri(ProcessContext context, ZendeskResource zendeskResource, ZendeskExportMethod exportMethod) {
String subDomain = context.getProperty(ZENDESK_SUBDOMAIN).evaluateAttributeExpressions().getValue();
String resourcePath = zendeskResource.apiPath(exportMethod);
HttpUriBuilder uriBuilder = uriBuilder(subDomain, resourcePath);
HttpUriBuilder uriBuilder = uriBuilder(resourcePath);
String cursor = getCursorState(context, zendeskResource, exportMethod);
if (cursor == null) {
@ -278,13 +206,6 @@ public class GetZendesk extends AbstractProcessor {
return uriBuilder.build();
}
HttpUriBuilder uriBuilder(String subDomain, String resourcePath) {
return webClientServiceProvider.getHttpUriBuilder()
.scheme(HTTPS)
.host(format(ZENDESK_HOST_TEMPLATE, subDomain))
.encodedPath(resourcePath);
}
private String getCursorState(ProcessContext context, ZendeskResource zendeskResource, ZendeskExportMethod exportMethod) {
try {
return context.getStateManager().getState(CLUSTER).get(zendeskResource.getValue() + exportMethod.getValue());
@ -293,23 +214,6 @@ public class GetZendesk extends AbstractProcessor {
}
}
private HttpResponseEntity performQuery(ProcessContext context, URI uri) {
String userName = context.getProperty(ZENDESK_USER).evaluateAttributeExpressions().getValue();
ZendeskAuthenticationType authenticationType = ZendeskAuthenticationType.forName(context.getProperty(ZENDESK_AUTHENTICATION_TYPE).getValue());
String authenticationCredential = context.getProperty(ZENDESK_AUTHENTICATION_CREDENTIAL).evaluateAttributeExpressions().getValue();
return webClientServiceProvider.getWebClientService()
.get()
.uri(uri)
.header(AUTHORIZATION_HEADER_NAME, basicAuthHeaderValue(authenticationType.enrichUserName(userName), authenticationCredential))
.retrieve();
}
private String basicAuthHeaderValue(String user, String credential) {
String userWithPassword = user + ":" + credential;
return BASIC_AUTH_PREFIX + getEncoder().encodeToString(userWithPassword.getBytes());
}
private OutputStreamCallback httpResponseParser(ProcessContext context, HttpResponseEntity response,
ZendeskResource zendeskResource, ZendeskExportMethod exportMethod,
AtomicInteger resultCount) {
@ -352,13 +256,4 @@ public class GetZendesk extends AbstractProcessor {
throw new ProcessException("Failed to update cursor state", e);
}
}
private String responseBodyToString(ProcessContext context, HttpResponseEntity response) {
try {
return IOUtils.toString(response.body(), UTF_8);
} catch (IOException e) {
context.yield();
throw new UncheckedIOException("Reading response body has failed", e);
}
}
}

268
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/main/java/org/apache/nifi/processors/zendesk/PutZendeskTicket.java Normal file
View File

@ -0,0 +1,268 @@
/*
* 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.zendesk;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.nifi.annotation.behavior.DynamicProperty;
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.common.zendesk.validation.JsonPointerPropertyNameValidator;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.web.client.api.HttpResponseEntity;
import org.apache.nifi.web.client.api.HttpUriBuilder;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.apache.nifi.common.zendesk.ZendeskProperties.APPLICATION_JSON;
import static org.apache.nifi.common.zendesk.ZendeskProperties.REL_FAILURE_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.WEB_CLIENT_SERVICE_PROVIDER;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_CREDENTIAL;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_TYPE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_CREATE_TICKETS_RESOURCE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_CREATE_TICKET_RESOURCE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_SUBDOMAIN;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_COMMENT_BODY;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_PRIORITY;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_SUBJECT;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_TYPE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_USER;
import static org.apache.nifi.common.zendesk.util.ZendeskRecordPathUtils.addDynamicField;
import static org.apache.nifi.common.zendesk.util.ZendeskRecordPathUtils.addField;
import static org.apache.nifi.common.zendesk.util.ZendeskUtils.createRequestObject;
import static org.apache.nifi.common.zendesk.util.ZendeskUtils.getDynamicProperties;
import static org.apache.nifi.common.zendesk.util.ZendeskUtils.getResponseBody;
import static org.apache.nifi.flowfile.attributes.CoreAttributes.MIME_TYPE;
import static org.apache.nifi.processors.zendesk.AbstractZendesk.RECORD_COUNT_ATTRIBUTE_NAME;
import static org.apache.nifi.processors.zendesk.PutZendeskTicket.ERROR_CODE_ATTRIBUTE_NAME;
import static org.apache.nifi.processors.zendesk.PutZendeskTicket.ERROR_MESSAGE_ATTRIBUTE_NAME;
import static org.apache.nifi.web.client.api.HttpResponseStatus.CREATED;
import static org.apache.nifi.web.client.api.HttpResponseStatus.OK;
@Tags({"zendesk, ticket"})
@CapabilityDescription("Create Zendesk tickets using the Zendesk API.")
@DynamicProperty(
name = "The path in the request object to add. The value needs be a valid JsonPointer.",
value = "The path in the incoming record to get the value from.",
expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
description = "Additional property to be added to the Zendesk request object.")
@WritesAttributes({
@WritesAttribute(attribute = RECORD_COUNT_ATTRIBUTE_NAME, description = "The number of records processed."),
@WritesAttribute(attribute = ERROR_CODE_ATTRIBUTE_NAME, description = "The error code of from the response."),
@WritesAttribute(attribute = ERROR_MESSAGE_ATTRIBUTE_NAME, description = "The error message of from the response.")})
public class PutZendeskTicket extends AbstractZendesk {
static final String ZENDESK_RECORD_READER_NAME = "zendesk-record-reader";
static final String ERROR_CODE_ATTRIBUTE_NAME = "error.code";
static final String ERROR_MESSAGE_ATTRIBUTE_NAME = "error.message";
private static final ObjectMapper mapper = new ObjectMapper();
static final PropertyDescriptor RECORD_READER = new PropertyDescriptor.Builder()
.name(ZENDESK_RECORD_READER_NAME)
.displayName("Record Reader")
.description("Specifies the Controller Service to use for parsing incoming data and determining the data's schema.")
.identifiesControllerService(RecordReaderFactory.class)
.build();
static final PropertyDescriptor TICKET_COMMENT_BODY = new PropertyDescriptor.Builder()
.fromPropertyDescriptor(ZENDESK_TICKET_COMMENT_BODY)
.dependsOn(RECORD_READER)
.build();
static final PropertyDescriptor TICKET_SUBJECT = new PropertyDescriptor.Builder()
.fromPropertyDescriptor(ZENDESK_TICKET_SUBJECT)
.dependsOn(RECORD_READER)
.build();
static final PropertyDescriptor TICKET_PRIORITY = new PropertyDescriptor.Builder()
.fromPropertyDescriptor(ZENDESK_TICKET_PRIORITY)
.dependsOn(RECORD_READER)
.build();
static final PropertyDescriptor TICKET_TYPE = new PropertyDescriptor.Builder()
.fromPropertyDescriptor(ZENDESK_TICKET_TYPE)
.dependsOn(RECORD_READER)
.build();
private static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
WEB_CLIENT_SERVICE_PROVIDER,
ZENDESK_SUBDOMAIN,
ZENDESK_USER,
ZENDESK_AUTHENTICATION_TYPE,
ZENDESK_AUTHENTICATION_CREDENTIAL,
RECORD_READER,
TICKET_COMMENT_BODY,
TICKET_SUBJECT,
TICKET_PRIORITY,
TICKET_TYPE
));
public static final Relationship REL_FAILURE = new Relationship.Builder()
.name(REL_FAILURE_NAME)
.description("A FlowFile is routed to this relationship if the operation failed and retrying the operation will also fail, such as an invalid data or schema.")
.build();
@Override
protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
return new PropertyDescriptor.Builder()
.name(propertyDescriptorName)
.required(false)
.addValidator(new JsonPointerPropertyNameValidator())
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.dynamic(true)
.build();
}
@Override
public Set<Relationship> getRelationships() {
return RELATIONSHIPS;
}
public static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
REL_SUCCESS,
REL_FAILURE
)));
@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return PROPERTIES;
}
@Override
public void onTrigger(ProcessContext context, ProcessSession session) {
FlowFile flowFile = session.get();
if (flowFile == null) {
return;
}
long startNanos = System.nanoTime();
HttpResponseEntity response;
URI uri;
final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
if (readerFactory == null) {
try (final InputStream inputStream = session.read(flowFile)) {
if (inputStream.available() == 0) {
inputStream.close();
getLogger().error("The incoming FlowFile's content is empty");
session.transfer(session.penalize(flowFile), REL_FAILURE);
return;
}
final HttpUriBuilder uriBuilder = uriBuilder(ZENDESK_CREATE_TICKET_RESOURCE);
uri = uriBuilder.build();
response = zendeskClient.performPostRequest(uri, inputStream);
} catch (IOException e) {
getLogger().error("Could not read the incoming FlowFile", e);
session.transfer(session.penalize(flowFile), REL_FAILURE);
return;
}
} else {
final String commentBody = context.getProperty(TICKET_COMMENT_BODY).evaluateAttributeExpressions().getValue();
final String subject = context.getProperty(TICKET_SUBJECT).evaluateAttributeExpressions().getValue();
final String priority = context.getProperty(TICKET_PRIORITY).evaluateAttributeExpressions().getValue();
final String type = context.getProperty(TICKET_TYPE).evaluateAttributeExpressions().getValue();
final Map<String, String> dynamicProperties = getDynamicProperties(context, context.getProperties(), flowFile.getAttributes());
List<ObjectNode> zendeskTickets = new ArrayList<>();
try (final InputStream in = session.read(flowFile); final RecordReader reader = readerFactory.createRecordReader(flowFile, in, getLogger())) {
Record record;
while ((record = reader.nextRecord()) != null) {
ObjectNode baseTicketNode = mapper.createObjectNode();
addField("/comment/body", commentBody, baseTicketNode, record);
addField("/subject", subject, baseTicketNode, record);
addField("/priority", priority, baseTicketNode, record);
addField("/type", type, baseTicketNode, record);
for (Map.Entry<String, String> dynamicProperty : dynamicProperties.entrySet()) {
addDynamicField(dynamicProperty.getKey(), dynamicProperty.getValue(), baseTicketNode, record);
}
zendeskTickets.add(baseTicketNode);
}
} catch (IOException | SchemaNotFoundException | MalformedRecordException e) {
getLogger().error("Error occurred while creating Zendesk tickets", e);
session.transfer(session.penalize(flowFile), REL_FAILURE);
return;
}
if (zendeskTickets.isEmpty()) {
getLogger().info("No records found in the incoming FlowFile");
flowFile = session.putAttribute(flowFile, RECORD_COUNT_ATTRIBUTE_NAME, "0");
session.transfer(flowFile, REL_SUCCESS);
return;
}
try {
final InputStream inputStream = createRequestObject(zendeskTickets);
uri = createUri(zendeskTickets.size());
response = zendeskClient.performPostRequest(uri, inputStream);
flowFile = session.putAttribute(flowFile, RECORD_COUNT_ATTRIBUTE_NAME, String.valueOf(zendeskTickets.size()));
} catch (IOException e) {
getLogger().error("Failed to post request to Zendesk", e);
session.transfer(session.penalize(flowFile), REL_FAILURE);
return;
}
}
handleResponse(session, flowFile, response, uri, startNanos);
}
private void handleResponse(ProcessSession session, FlowFile flowFile, HttpResponseEntity response, URI uri, long startNanos) {
if (response.statusCode() == CREATED.getCode() || response.statusCode() == OK.getCode()) {
flowFile = session.putAttribute(flowFile, MIME_TYPE.key(), APPLICATION_JSON);
session.transfer(flowFile, REL_SUCCESS);
long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
session.getProvenanceReporter().send(flowFile, uri.toString(), transferMillis);
} else {
String errorMessage = getResponseBody(response);
getLogger().error("Zendesk ticket creation returned with error, HTTP status={}, response={}", response.statusCode(), errorMessage);
flowFile = session.putAttribute(flowFile, ERROR_CODE_ATTRIBUTE_NAME, String.valueOf(response.statusCode()));
flowFile = session.putAttribute(flowFile, ERROR_MESSAGE_ATTRIBUTE_NAME, errorMessage);
session.transfer(session.penalize(flowFile), REL_FAILURE);
}
}
private URI createUri(int numberOfTickets) {
final String resource = numberOfTickets > 1 ? ZENDESK_CREATE_TICKETS_RESOURCE : ZENDESK_CREATE_TICKET_RESOURCE;
return uriBuilder(resource).build();
}
}

3
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
View File

@ -12,4 +12,5 @@
# 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.
org.apache.nifi.processors.zendesk.GetZendesk
org.apache.nifi.processors.zendesk.GetZendesk
org.apache.nifi.processors.zendesk.PutZendeskTicket

166
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/main/resources/docs/org.apache.nifi.processors.zendesk.PutZendeskTicket/additionalDetails.html Normal file
View File

@ -0,0 +1,166 @@
<!DOCTYPE html>
<html lang="en" xmlns="http://www.w3.org/1999/html">
<!--
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.
-->
<head>
<meta charset="utf-8"/>
<title>PutZendeskTicket</title>
<link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
<style>
h2 {margin-top: 4em}
h3 {margin-top: 3em}
</style>
</head>
<body>
<h1>PutZendeskTicket</h1>
<h3>Description</h3>
<p>
The processor uses the Zendesk API to ingest tickets into Zendesk. The processor is capable to send requests directly from the FlowFile content or construct the request objects from the incoming records using a RecordReader.
</p>
<h3>Authentication</h3>
<p>
Zendesk API uses basic authentication. Either a password or an authentication token has to be provided.
In Zendesk API Settings, it's possible to generate authentication tokens, eliminating the need for users to expose their passwords. This approach also offers the advantage of fast token revocation when required.
</p>
<h3>Property values</h3>
<p>
There are multiple ways of providing property values to the request object:
<dl>
<dt><b>Record Path:</b></dt>
<dd>
<p>The property value is going to be evaluated as a record path if the value is provided inside brackets starting with a '%'. </p>
<p><u>Example:</u></p>
<p>
The incoming record look like this.
<pre>
{
"record":{
"description":"This is a sample description.",
"issue_type":"Immediate",
"issue":{
"name":"General error",
"type":"Immediate"
},
"project":{
"name":"Maintenance"
}
}
}
</pre>
We are going to provide Record Path values for the <em>Comment Body, Subject, Priority</em> and <em>Type</em> processor attributes:
<pre>
Comment Body : %{/record/description}
Subject : %{/record/issue/name}
Priority : %{/record/issue/type}
Type : %{/record/project/name}
</pre>
The constructed request object that is going to be sent to the Zendesk API will look like this:
<pre>
{
"comment":{
"body":"This is a sample description."
},
"subject":"General error",
"priority":"Immediate",
"type":"Maintenance"
}
</pre>
</p>
</dd>
<dt><b>Constant:</b></dt>
<dd>
<p>The property value is going to be treated as a constant if the provided value doesn't match with the Record Path format.</p>
<p><u>Example:</u></p>
<p>
We are going to provide constant values for the <em>Comment Body, Subject, Priority</em> and <em>Type</em> processor attributes:
<pre>
Comment Body : Sample description
Subject : Sample subject
Priority : High
Type : Sample type
</pre>
The constructed request object that is going to be sent to the Zendesk API will look like this:
<pre>
{
"comment":{
"body":"Sample description"
},
"subject":"Sample subject",
"priority":"High",
"type":"Sample type"
}
</pre>
</p>
</dd>
</dl>
</p>
<h3>Additional properties</h3>
<p>
The processor offers a set of frequently used Zendesk ticket attributes within its property list. However, users have the flexibility to include any desired number of additional properties using dynamic properties.
These dynamic properties utilize their keys as Json Pointer, which denote the paths within the request object. Correspondingly, the values of these dynamic properties align with the predefined property attributes.
The possible Zendesk request attributes can be found in the <a target="blank" href="https://developer.zendesk.com/api-reference/ticketing/tickets/tickets/">Zendesk API documentation</a>
</p>
<p><u>Property Key values:</u></p>
<p>
The dynamic property key must be a valid Json Pointer value which has the following syntax rules:
<ul>
<li>The path starts with <b>/</b>.</li>
<li>Each segment is separated by <b>/</b>.</li>
<li>Each segment can be interpreted as either an array index or an object key.</li>
</ul>
</p>
<p><u>Example:</u></p>
<p>
We are going to add a new dynamic property to the processor:
<pre>
/request/new_object : This is a new property
/request/new_array/0 : This is a new array element
</pre>
The constructed request object will look like this:
<pre>
{
"request":{
"new_object":"This is a new property",
"new_array":[
"This is a new array element"
]
}
}
</pre>
</p>
</body>
</html>

56
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/test/java/org/apache/nifi/processors/zendesk/GetZendeskTest.java
View File

@ -17,36 +17,11 @@
package org.apache.nifi.processors.zendesk;
import static java.lang.String.format;
import static java.util.Arrays.asList;
import static java.util.Collections.singletonMap;
import static org.apache.nifi.components.state.Scope.CLUSTER;
import static org.apache.nifi.processors.zendesk.GetZendesk.HTTP_TOO_MANY_REQUESTS;
import static org.apache.nifi.processors.zendesk.GetZendesk.RECORD_COUNT_ATTRIBUTE_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.REL_SUCCESS_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.WEB_CLIENT_SERVICE_PROVIDER_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_AUTHENTICATION_CREDENTIAL_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_AUTHENTICATION_TYPE_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_EXPORT_METHOD_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_QUERY_START_TIMESTAMP_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_RESOURCE_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_SUBDOMAIN_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_USER_NAME;
import static org.apache.nifi.processors.zendesk.ZendeskExportMethod.CURSOR;
import static org.apache.nifi.processors.zendesk.ZendeskResource.TICKETS;
import static org.apache.nifi.util.TestRunners.newTestRunner;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;
import okhttp3.HttpUrl;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import okhttp3.mockwebserver.RecordedRequest;
import org.apache.nifi.common.zendesk.ZendeskAuthenticationType;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
@ -61,6 +36,33 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.opentest4j.AssertionFailedError;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;
import static java.lang.String.format;
import static java.util.Arrays.asList;
import static java.util.Collections.singletonMap;
import static org.apache.nifi.common.zendesk.ZendeskProperties.REL_SUCCESS_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.WEB_CLIENT_SERVICE_PROVIDER_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_CREDENTIAL_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_TYPE_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_SUBDOMAIN_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_USER_NAME;
import static org.apache.nifi.components.state.Scope.CLUSTER;
import static org.apache.nifi.processors.zendesk.GetZendesk.HTTP_TOO_MANY_REQUESTS;
import static org.apache.nifi.processors.zendesk.GetZendesk.RECORD_COUNT_ATTRIBUTE_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_EXPORT_METHOD_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_QUERY_START_TIMESTAMP_NAME;
import static org.apache.nifi.processors.zendesk.GetZendesk.ZENDESK_RESOURCE_NAME;
import static org.apache.nifi.processors.zendesk.ZendeskExportMethod.CURSOR;
import static org.apache.nifi.processors.zendesk.ZendeskResource.TICKETS;
import static org.apache.nifi.util.TestRunners.newTestRunner;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class GetZendeskTest {
private static final int HTTP_OK = 200;
@ -255,7 +257,7 @@ public class GetZendeskTest {
class TestGetZendesk extends GetZendesk {
@Override
HttpUriBuilder uriBuilder(String subDomain, String resourcePath) {
HttpUriBuilder uriBuilder(String resourcePath) {
HttpUrl url = server.url(resourcePath);
return new StandardHttpUriBuilder()
.scheme(url.scheme())

366
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/test/java/org/apache/nifi/processors/zendesk/PutZendeskTicketTest.java Normal file
View File

@ -0,0 +1,366 @@
/*
* 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.zendesk;
import com.fasterxml.jackson.databind.ObjectMapper;
import okhttp3.HttpUrl;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import okhttp3.mockwebserver.RecordedRequest;
import org.apache.nifi.common.zendesk.ZendeskAuthenticationType;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.serialization.record.MockRecordParser;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.web.client.StandardHttpUriBuilder;
import org.apache.nifi.web.client.api.HttpUriBuilder;
import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
import org.apache.nifi.web.client.provider.service.StandardWebClientServiceProvider;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import static org.apache.nifi.common.zendesk.ZendeskProperties.WEB_CLIENT_SERVICE_PROVIDER_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_CREDENTIAL_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_TYPE_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_CREATE_TICKETS_RESOURCE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_CREATE_TICKET_RESOURCE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_SUBDOMAIN_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_COMMENT_BODY_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_PRIORITY_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_SUBJECT_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_TYPE_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_USER_NAME;
import static org.apache.nifi.processors.zendesk.AbstractZendesk.RECORD_COUNT_ATTRIBUTE_NAME;
import static org.apache.nifi.processors.zendesk.PutZendeskTicket.REL_FAILURE;
import static org.apache.nifi.processors.zendesk.PutZendeskTicket.REL_SUCCESS;
import static org.apache.nifi.processors.zendesk.PutZendeskTicket.ZENDESK_RECORD_READER_NAME;
import static org.apache.nifi.util.TestRunners.newTestRunner;
import static org.junit.jupiter.api.Assertions.assertEquals;
public class PutZendeskTicketTest {
private static final int HTTP_OK = 200;
private static final int HTTP_BAD_REQUEST = 400;
private static final String EMPTY_RESPONSE = "{}";
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private MockWebServer server;
private TestRunner testRunner;
@BeforeEach
public void init() throws IOException, InitializationException {
server = new MockWebServer();
server.start();
testRunner = newTestRunner(new TestPutZendeskTicket());
WebClientServiceProvider webClientServiceProvider = new StandardWebClientServiceProvider();
testRunner.addControllerService("web-client-service-provider", webClientServiceProvider);
testRunner.enableControllerService(webClientServiceProvider);
testRunner.setProperty(WEB_CLIENT_SERVICE_PROVIDER_NAME, "web-client-service-provider");
testRunner.setProperty(ZENDESK_SUBDOMAIN_NAME, "default-zendesk-subdomain");
testRunner.setProperty(ZENDESK_USER_NAME, "default-zendesk-user-name");
testRunner.setProperty(ZENDESK_AUTHENTICATION_TYPE_NAME, ZendeskAuthenticationType.PASSWORD.getValue());
testRunner.setProperty(ZENDESK_AUTHENTICATION_CREDENTIAL_NAME, "default-zendesk-password");
}
@AfterEach
void tearDown() throws IOException {
server.shutdown();
}
@Test
public void testOnTriggerWithoutRecordReader() throws InterruptedException, IOException {
String flowFileContent =
"{\n" +
" \"ticket\" : {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test description\"\n" +
" },\n" +
" \"subject\" : \"Test subject\",\n" +
" \"priority\" : \"High\",\n" +
" \"type\" : \"Development\"\n" +
" }\n" +
"}";
MockFlowFile flowFile = new MockFlowFile(1L);
flowFile.setData(flowFileContent.getBytes());
// given
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
// when
testRunner.enqueue(flowFile);
testRunner.run();
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKET_RESOURCE, recordedRequest.getPath());
assertEquals(OBJECT_MAPPER.readTree(flowFileContent), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
testRunner.assertAllFlowFilesTransferred(REL_SUCCESS);
}
@Test
public void testOnTriggerWithFixPropertiesAndSingleTicket() throws InterruptedException, InitializationException, IOException {
MockRecordParser reader = new MockRecordParser();
reader.addSchemaField("description", RecordFieldType.STRING);
reader.addSchemaField("subject", RecordFieldType.STRING);
reader.addSchemaField("priority", RecordFieldType.STRING);
reader.addSchemaField("type", RecordFieldType.STRING);
reader.addRecord("This is a test description", "Test subject", "High", "Development");
testRunner.addControllerService("mock-reader-factory", reader);
testRunner.enableControllerService(reader);
testRunner.setProperty(ZENDESK_RECORD_READER_NAME, "mock-reader-factory");
// given
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
testRunner.setProperty(ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
testRunner.setProperty(ZENDESK_TICKET_SUBJECT_NAME, "@{/subject}");
testRunner.setProperty(ZENDESK_TICKET_PRIORITY_NAME, "@{/priority}");
testRunner.setProperty(ZENDESK_TICKET_TYPE_NAME, "@{/type}");
// when
testRunner.enqueue(new byte[0]);
testRunner.run(1);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKET_RESOURCE, recordedRequest.getPath());
String expectedBody =
"{\n" +
" \"ticket\" : {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test description\"\n" +
" },\n" +
" \"subject\" : \"Test subject\",\n" +
" \"priority\" : \"High\",\n" +
" \"type\" : \"Development\"\n" +
" }\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
testRunner.assertAllFlowFilesTransferred(REL_SUCCESS);
}
@Test
public void testOnTriggerWithFixPropertiesAndMultipleTickets() throws InterruptedException, InitializationException, IOException {
MockRecordParser reader = new MockRecordParser();
reader.addSchemaField("description", RecordFieldType.STRING);
reader.addRecord("This is a test description1");
reader.addRecord("This is a test description2");
testRunner.addControllerService("mock-reader-factory", reader);
testRunner.enableControllerService(reader);
testRunner.setProperty(ZENDESK_RECORD_READER_NAME, "mock-reader-factory");
// given
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
testRunner.setProperty(ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
// when
testRunner.enqueue(new byte[0]);
testRunner.run(1);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKETS_RESOURCE, recordedRequest.getPath());
String expectedBody =
"{\n" +
" \"tickets\" : [ {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test description1\"\n" +
" }\n" +
" }, {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test description2\"\n" +
" }\n" +
" } ]\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
testRunner.assertAllFlowFilesTransferred(REL_SUCCESS);
}
@Test
public void testOnTriggerWithRecordPathDynamicProperties() throws InterruptedException, InitializationException, IOException {
MockRecordParser reader = new MockRecordParser();
reader.addSchemaField("description", RecordFieldType.STRING);
reader.addSchemaField("dynamicPropertySource1", RecordFieldType.STRING);
reader.addSchemaField("dynamicPropertySource2", RecordFieldType.STRING);
reader.addRecord("This is a test description", "This is a dynamic property 1", "This is a dynamic property 2");
testRunner.addControllerService("mock-reader-factory", reader);
testRunner.enableControllerService(reader);
testRunner.setProperty(ZENDESK_RECORD_READER_NAME, "mock-reader-factory");
// given
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
testRunner.setProperty(ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
testRunner.setProperty("/dp1/dynamicPropertyTarget1", "@{/dynamicPropertySource1}");
testRunner.setProperty("/dp1/dp2/dp3/dynamicPropertyTarget2", "@{/dynamicPropertySource2}");
// when
testRunner.enqueue(new byte[0]);
testRunner.run(1);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKET_RESOURCE, recordedRequest.getPath());
String expectedBody =
"{\n" +
" \"ticket\" : {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test description\"\n" +
" },\n" +
" \"dp1\" : {\n" +
" \"dp2\" : {\n" +
" \"dp3\" : {\n" +
" \"dynamicPropertyTarget2\" : \"This is a dynamic property 2\"\n" +
" }\n" +
" },\n" +
" \"dynamicPropertyTarget1\" : \"This is a dynamic property 1\"\n" +
" }\n" +
" }\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
testRunner.assertAllFlowFilesTransferred(REL_SUCCESS);
}
@Test
public void testOnTriggerWithConstantDynamicProperties() throws InterruptedException, InitializationException, IOException {
MockRecordParser reader = new MockRecordParser();
reader.addSchemaField("description", RecordFieldType.STRING);
reader.addSchemaField("dynamicPropertySource1", RecordFieldType.STRING);
reader.addSchemaField("dynamicPropertySource2", RecordFieldType.STRING);
reader.addRecord("This is a test description", "This is a dynamic property 1", "This is a dynamic property 2");
testRunner.addControllerService("mock-reader-factory", reader);
testRunner.enableControllerService(reader);
testRunner.setProperty(ZENDESK_RECORD_READER_NAME, "mock-reader-factory");
// given
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
testRunner.setProperty(ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
testRunner.setProperty("/dp1/dynamicPropertyTarget1", "Constant 1");
testRunner.setProperty("/dp1/dp2/dp3/dynamicPropertyTarget2", "Constant2");
// when
testRunner.enqueue(new byte[0]);
testRunner.run(1);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKET_RESOURCE, recordedRequest.getPath());
String expectedBody =
"{\n" +
" \"ticket\" : {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test description\"\n" +
" },\n" +
" \"dp1\" : {\n" +
" \"dp2\" : {\n" +
" \"dp3\" : {\n" +
" \"dynamicPropertyTarget2\" : \"Constant2\"\n" +
" }\n" +
" },\n" +
" \"dynamicPropertyTarget1\" : \"Constant 1\"\n" +
" }\n" +
" }\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
testRunner.assertAllFlowFilesTransferred(REL_SUCCESS);
}
@Test
public void testOnTriggerWithErrorResponse() throws InitializationException {
MockRecordParser reader = new MockRecordParser();
reader.addSchemaField("description", RecordFieldType.STRING);
reader.addRecord("This is a test description");
testRunner.addControllerService("mock-reader-factory", reader);
testRunner.enableControllerService(reader);
testRunner.setProperty(ZENDESK_RECORD_READER_NAME, "mock-reader-factory");
// given
server.enqueue(new MockResponse().setResponseCode(HTTP_BAD_REQUEST).setBody(EMPTY_RESPONSE));
testRunner.setProperty(ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
// when
testRunner.enqueue(new byte[0]);
testRunner.run(1);
// then
testRunner.assertAllFlowFilesTransferred(REL_FAILURE);
}
@Test
public void testOnTriggerWithZeroRecord() throws InitializationException {
MockRecordParser reader = new MockRecordParser();
testRunner.addControllerService("mock-reader-factory", reader);
testRunner.enableControllerService(reader);
testRunner.setProperty(ZENDESK_RECORD_READER_NAME, "mock-reader-factory");
testRunner.setProperty(ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
// when
testRunner.enqueue(new byte[0]);
testRunner.run();
// then
testRunner.assertTransferCount(REL_SUCCESS, 1);
MockFlowFile flowFile = testRunner.getFlowFilesForRelationship(REL_SUCCESS).get(0);
assertEquals("0", flowFile.getAttribute(RECORD_COUNT_ATTRIBUTE_NAME));
}
@Test
public void testOnTriggerWithEmptyFlowFileWithoutRecordReader() {
MockFlowFile flowFile = new MockFlowFile(1L);
flowFile.setData("".getBytes());
// when
testRunner.enqueue(flowFile);
testRunner.run();
// then
testRunner.assertTransferCount(REL_FAILURE, 1);
}
class TestPutZendeskTicket extends PutZendeskTicket {
@Override
HttpUriBuilder uriBuilder(String resourcePath) {
HttpUrl url = server.url(resourcePath);
return new StandardHttpUriBuilder()
.scheme(url.scheme())
.host(url.host())
.port(url.port())
.encodedPath(url.encodedPath());
}
}
}

2
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-processors/src/test/java/org/apache/nifi/processors/zendesk/ZendeskAuthenticationTypeTest.java
View File

@ -20,6 +20,8 @@ package org.apache.nifi.processors.zendesk;
import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.stream.Stream;
import org.apache.nifi.common.zendesk.ZendeskAuthenticationType;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;

43
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-services-nar/pom.xml Normal file
View File

@ -0,0 +1,43 @@
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-zendesk-bundle</artifactId>
<version>2.0.0-SNAPSHOT</version>
</parent>
<artifactId>nifi-zendesk-services-nar</artifactId>
<packaging>nar</packaging>
<dependencies>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-zendesk-services</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-standard-shared-nar</artifactId>
<type>nar</type>
</dependency>
</dependencies>
</project>

240
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-services-nar/src/main/resources/META-INF/LICENSE Normal file
View File

@ -0,0 +1,240 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed 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.
APACHE NIFI SUBCOMPONENTS:
The Apache NiFi project contains subcomponents with separate copyright
notices and license terms. Your use of the source code for the these
subcomponents is subject to the terms and conditions of the following
licenses.
The binary distribution of this product bundles 'Antlr 3' which is available
under a "3-clause BSD" license. For details see http://www.antlr3.org/license.html
Copyright (c) 2010 Terence Parr
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
Neither the name of the author nor the names of its contributors may be used
to endorse or promote products derived from this software without specific
prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
THE POSSIBILITY OF SUCH DAMAGE.

5
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-services-nar/src/main/resources/META-INF/NOTICE Normal file
View File

@ -0,0 +1,5 @@
nifi-zendesk-services-nar
Copyright 2015-2022 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).

85
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-services/pom.xml Normal file
View File

@ -0,0 +1,85 @@
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-zendesk-bundle</artifactId>
<version>2.0.0-SNAPSHOT</version>
</parent>
<artifactId>nifi-zendesk-services</artifactId>
<packaging>jar</packaging>
<dependencies>
<!-- Internal dependencies -->
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-zendesk-common</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-record-sink-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-record-serialization-service-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-web-client-provider-api</artifactId>
<scope>provided</scope>
</dependency>
<!-- Test dependencies -->
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-mock</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-web-client-provider-service</artifactId>
<version>2.0.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-ssl-context-service-api</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-proxy-configuration-api</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>mockwebserver</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-mock-record-utils</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

215
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-services/src/main/java/org/apache/nifi/services/zendesk/ZendeskRecordSink.java Normal file
View File

@ -0,0 +1,215 @@
/*
* 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.services.zendesk;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnDisabled;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.common.zendesk.ZendeskAuthenticationContext;
import org.apache.nifi.common.zendesk.ZendeskAuthenticationType;
import org.apache.nifi.common.zendesk.ZendeskClient;
import org.apache.nifi.common.zendesk.validation.JsonPointerPropertyNameValidator;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.record.sink.RecordSinkService;
import org.apache.nifi.serialization.WriteResult;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordSet;
import org.apache.nifi.web.client.api.HttpResponseEntity;
import org.apache.nifi.web.client.api.HttpResponseStatus;
import org.apache.nifi.web.client.api.HttpUriBuilder;
import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import static org.apache.nifi.common.zendesk.ZendeskProperties.WEB_CLIENT_SERVICE_PROVIDER;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_CREDENTIAL;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_TYPE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_CREATE_TICKETS_RESOURCE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_CREATE_TICKET_RESOURCE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_SUBDOMAIN;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_COMMENT_BODY;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_PRIORITY;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_SUBJECT;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_TYPE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_USER;
import static org.apache.nifi.common.zendesk.util.ZendeskRecordPathUtils.addDynamicField;
import static org.apache.nifi.common.zendesk.util.ZendeskRecordPathUtils.addField;
import static org.apache.nifi.common.zendesk.util.ZendeskUtils.createRequestObject;
import static org.apache.nifi.common.zendesk.util.ZendeskUtils.getDynamicProperties;
import static org.apache.nifi.common.zendesk.util.ZendeskUtils.getResponseBody;
@Tags({"zendesk", "record", "sink"})
@CapabilityDescription("Create Zendesk tickets using the Zendesk API." +
"The service requires a Zendesk account with configured access.")
public class ZendeskRecordSink extends AbstractControllerService implements RecordSinkService {
private final ObjectMapper mapper = new ObjectMapper();
private Map<String, String> dynamicProperties;
private Cache<String, ObjectNode> recordCache;
private ZendeskClient zendeskClient;
private String commentBody;
private String subject;
private String priority;
private String type;
static final PropertyDescriptor CACHE_SIZE = new PropertyDescriptor.Builder()
.name("cache-size")
.displayName("Cache Size")
.description("Specifies how many Zendesk ticket should be cached.")
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
.defaultValue("1000")
.required(true)
.build();
static final PropertyDescriptor CACHE_EXPIRATION = new PropertyDescriptor.Builder()
.name("cache-expiration")
.displayName("Cache Expiration")
.description("Specifies how long a Zendesk ticket that is cached should remain in the cache.")
.addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
.defaultValue("1 hour")
.required(true)
.build();
private static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
WEB_CLIENT_SERVICE_PROVIDER,
ZENDESK_SUBDOMAIN,
ZENDESK_USER,
ZENDESK_AUTHENTICATION_TYPE,
ZENDESK_AUTHENTICATION_CREDENTIAL,
ZENDESK_TICKET_COMMENT_BODY,
ZENDESK_TICKET_SUBJECT,
ZENDESK_TICKET_PRIORITY,
ZENDESK_TICKET_TYPE,
CACHE_SIZE,
CACHE_EXPIRATION
));
@Override
protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
return new PropertyDescriptor.Builder()
.name(propertyDescriptorName)
.required(false)
.addValidator(new JsonPointerPropertyNameValidator())
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.dynamic(true)
.build();
}
@Override
public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return PROPERTIES;
}
@Override
public WriteResult sendData(RecordSet recordSet, Map<String, String> attributes, boolean sendZeroResults) throws IOException {
List<ObjectNode> zendeskTickets = new ArrayList<>();
Record record;
while ((record = recordSet.next()) != null) {
ObjectNode baseTicketNode = mapper.createObjectNode();
addField("/comment/body", commentBody, baseTicketNode, record);
addField("/subject", subject, baseTicketNode, record);
addField("/priority", priority, baseTicketNode, record);
addField("/type", type, baseTicketNode, record);
for (Map.Entry<String, String> dynamicProperty : dynamicProperties.entrySet()) {
addDynamicField(dynamicProperty.getKey(), dynamicProperty.getValue(), baseTicketNode, record);
}
ObjectNode ticketNode = recordCache.getIfPresent(baseTicketNode.toString());
if (ticketNode == null) {
recordCache.put(baseTicketNode.toString(), baseTicketNode);
zendeskTickets.add(baseTicketNode);
}
}
if (!zendeskTickets.isEmpty()) {
try {
final InputStream inputStream = createRequestObject(zendeskTickets);
final URI uri = createUri(zendeskTickets.size());
final HttpResponseEntity response = zendeskClient.performPostRequest(uri, inputStream);
if (response.statusCode() != HttpResponseStatus.CREATED.getCode() && response.statusCode() != HttpResponseStatus.OK.getCode()) {
getLogger().error("Failed to create zendesk ticket, HTTP status={}, response={}", response.statusCode(), getResponseBody(response));
}
} catch (IOException e) {
throw new IOException("Failed to post request to Zendesk", e);
}
}
return WriteResult.of(zendeskTickets.size(), Collections.emptyMap());
}
@OnEnabled
public void onEnabled(final ConfigurationContext context) {
dynamicProperties = getDynamicProperties(context, context.getProperties(), Collections.emptyMap());
commentBody = context.getProperty(ZENDESK_TICKET_COMMENT_BODY).evaluateAttributeExpressions().getValue();
subject = context.getProperty(ZENDESK_TICKET_SUBJECT).evaluateAttributeExpressions().getValue();
priority = context.getProperty(ZENDESK_TICKET_PRIORITY).evaluateAttributeExpressions().getValue();
type = context.getProperty(ZENDESK_TICKET_TYPE).evaluateAttributeExpressions().getValue();
final String subdomain = context.getProperty(ZENDESK_SUBDOMAIN).evaluateAttributeExpressions().getValue();
final String user = context.getProperty(ZENDESK_USER).evaluateAttributeExpressions().getValue();
final ZendeskAuthenticationType authenticationType = ZendeskAuthenticationType.forName(context.getProperty(ZENDESK_AUTHENTICATION_TYPE).getValue());
final String authenticationCredentials = context.getProperty(ZENDESK_AUTHENTICATION_CREDENTIAL).evaluateAttributeExpressions().getValue();
final ZendeskAuthenticationContext authenticationContext = new ZendeskAuthenticationContext(subdomain, user, authenticationType, authenticationCredentials);
final WebClientServiceProvider webClientServiceProvider = context.getProperty(WEB_CLIENT_SERVICE_PROVIDER).asControllerService(WebClientServiceProvider.class);
zendeskClient = new ZendeskClient(webClientServiceProvider, authenticationContext);
final int cacheSize = context.getProperty(CACHE_SIZE).asInteger();
final long cacheExpiration = context.getProperty(CACHE_EXPIRATION).asTimePeriod(TimeUnit.NANOSECONDS);
recordCache = Caffeine.newBuilder()
.maximumSize(cacheSize)
.expireAfterWrite(Duration.ofNanos(cacheExpiration))
.build();
}
@OnDisabled
public void onDisabled() {
recordCache.invalidateAll();
}
private URI createUri(int numberOfTickets) {
final String resource = numberOfTickets > 1 ? ZENDESK_CREATE_TICKETS_RESOURCE : ZENDESK_CREATE_TICKET_RESOURCE;
return uriBuilder(resource).build();
}
HttpUriBuilder uriBuilder(String resourcePath) {
return zendeskClient.uriBuilder(resourcePath);
}
}

15
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService Normal file
View File

@ -0,0 +1,15 @@
# 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.
org.apache.nifi.services.zendesk.ZendeskRecordSink

172
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-services/src/main/resources/docs/org.apache.nifi.services.zendesk.ZendeskRecordSink/additionalDetails.html Normal file
View File

@ -0,0 +1,172 @@
<!DOCTYPE html>
<html lang="en" xmlns="http://www.w3.org/1999/html">
<!--
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.
-->
<head>
<meta charset="utf-8"/>
<title>ZendeskRecordSink</title>
<link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
<style>
h2 {margin-top: 4em}
h3 {margin-top: 3em}
</style>
</head>
<body>
<h1>ZendeskRecordSink</h1>
<h3>Description</h3>
<p>
The sink uses the Zendesk API to ingest tickets into Zendesk, using the incoming records to construct request objects.
</p>
<h3>Authentication</h3>
<p>
Zendesk API uses basic authentication. Either a password or an authentication token has to be provided.
In Zendesk API Settings, it's possible to generate authentication tokens, eliminating the need for users to expose their passwords. This approach also offers the advantage of fast token revocation when required.
</p>
<h3>Property values</h3>
<p>
There are multiple ways of providing property values to the request object:
<dl>
<dt><b>Record Path:</b></dt>
<dd>
<p>The property value is going to be evaluated as a record path if the value is provided inside brackets starting with a '%'. </p>
<p><u>Example:</u></p>
<p>
The incoming record look like this.
<pre>
{
"record":{
"description":"This is a sample description.",
"issue_type":"Immediate",
"issue":{
"name":"General error",
"type":"Immediate"
},
"project":{
"name":"Maintenance"
}
}
}
</pre>
We are going to provide Record Path values for the <em>Comment Body, Subject, Priority</em> and <em>Type</em> processor attributes:
<pre>
Comment Body : %{/record/description}
Subject : %{/record/issue/name}
Priority : %{/record/issue/type}
Type : %{/record/project/name}
</pre>
The constructed request object that is going to be sent to the Zendesk API will look like this:
<pre>
{
"comment":{
"body":"This is a sample description."
},
"subject":"General error",
"priority":"Immediate",
"type":"Maintenance"
}
</pre>
</p>
</dd>
<dt><b>Constant:</b></dt>
<dd>
<p>The property value is going to be treated as a constant if the provided value doesn't match with the Record Path format.</p>
<p><u>Example:</u></p>
<p>
We are going to provide constant values for the <em>Comment Body, Subject, Priority</em> and <em>Type</em> processor attributes:
<pre>
Comment Body : Sample description
Subject : Sample subject
Priority : High
Type : Sample type
</pre>
The constructed request object that is going to be sent to the Zendesk API will look like this:
<pre>
{
"comment":{
"body":"Sample description"
},
"subject":"Sample subject",
"priority":"High",
"type":"Sample type"
}
</pre>
</p>
</dd>
</dl>
</p>
<h3>Additional properties</h3>
<p>
The processor offers a set of frequently used Zendesk ticket attributes within its property list. However, users have the flexibility to include any desired number of additional properties using dynamic properties.
These dynamic properties utilize their keys as Json Pointer, which denote the paths within the request object. Correspondingly, the values of these dynamic properties align with the predefined property attributes.
The possible Zendesk request attributes can be found in the <a target="blank" href="https://developer.zendesk.com/api-reference/ticketing/tickets/tickets/">Zendesk API documentation</a>
</p>
<p><u>Property Key values:</u></p>
<p>
The dynamic property key must be a valid Json Pointer value which has the following syntax rules:
<ul>
<li>The path starts with <b>/</b>.</li>
<li>Each segment is separated by <b>/</b>.</li>
<li>Each segment can be interpreted as either an array index or an object key.</li>
</ul>
</p>
<p><u>Example:</u></p>
<p>
We are going to add a new dynamic property to the processor:
<pre>
/request/new_object : This is a new property
/request/new_array/0 : This is a new array element
</pre>
The constructed request object will look like this:
<pre>
{
"request":{
"new_object":"This is a new property",
"new_array":[
"This is a new array element"
]
}
}
</pre>
</p>
<h3>Caching</h3>
<p>
The sink caches Zendesk tickets with the same content in order to avoid duplicate issues.
The cache size and expiration date can be set on the sink service.
</p>
</body>
</html>

365
nifi-nar-bundles/nifi-zendesk-bundle/nifi-zendesk-services/src/test/java/org/apache/nifi/services/zendesk/ZendeskRecordSinkTest.java Normal file
View File

@ -0,0 +1,365 @@
/*
* 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.services.zendesk;
import com.fasterxml.jackson.databind.ObjectMapper;
import okhttp3.HttpUrl;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import okhttp3.mockwebserver.RecordedRequest;
import org.apache.nifi.common.zendesk.ZendeskAuthenticationType;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.WriteResult;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.RecordSet;
import org.apache.nifi.util.NoOpProcessor;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.apache.nifi.web.client.StandardHttpUriBuilder;
import org.apache.nifi.web.client.api.HttpUriBuilder;
import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
import org.apache.nifi.web.client.provider.service.StandardWebClientServiceProvider;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.apache.nifi.common.zendesk.ZendeskProperties.WEB_CLIENT_SERVICE_PROVIDER_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_CREDENTIAL_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_AUTHENTICATION_TYPE_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_CREATE_TICKETS_RESOURCE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_CREATE_TICKET_RESOURCE;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_SUBDOMAIN_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_COMMENT_BODY_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_PRIORITY_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_SUBJECT_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_TICKET_TYPE_NAME;
import static org.apache.nifi.common.zendesk.ZendeskProperties.ZENDESK_USER_NAME;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
public class ZendeskRecordSinkTest {
private static final int HTTP_OK = 200;
private static final String EMPTY_RESPONSE = "{}";
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private MockWebServer server;
private TestRunner testRunner;
private RecordSet recordSet;
private TestZendeskRecordSink sinkZendeskTicket;
@BeforeEach
public void init() throws IOException, InitializationException {
server = new MockWebServer();
server.start();
testRunner = TestRunners.newTestRunner(new NoOpProcessor());
sinkZendeskTicket = new TestZendeskRecordSink();
WebClientServiceProvider webClientServiceProvider = new StandardWebClientServiceProvider();
testRunner.addControllerService("web-client-service-provider", webClientServiceProvider);
testRunner.enableControllerService(webClientServiceProvider);
testRunner.addControllerService("sinkZendeskTicket", sinkZendeskTicket);
testRunner.setProperty(sinkZendeskTicket, WEB_CLIENT_SERVICE_PROVIDER_NAME, "web-client-service-provider");
testRunner.setProperty(sinkZendeskTicket, ZENDESK_SUBDOMAIN_NAME, "default-zendesk-subdomain");
testRunner.setProperty(sinkZendeskTicket, ZENDESK_USER_NAME, "default-zendesk-user-name");
testRunner.setProperty(sinkZendeskTicket, ZENDESK_AUTHENTICATION_TYPE_NAME, ZendeskAuthenticationType.PASSWORD.getValue());
testRunner.setProperty(sinkZendeskTicket, ZENDESK_AUTHENTICATION_CREDENTIAL_NAME, "default-zendesk-password");
}
@AfterEach
void tearDown() throws IOException {
server.shutdown();
}
private void initSingleTestRecord() {
List<RecordField> fields = new ArrayList<>();
fields.add(new RecordField("description", RecordFieldType.STRING.getDataType()));
fields.add(new RecordField("subject", RecordFieldType.STRING.getDataType()));
fields.add(new RecordField("priority", RecordFieldType.STRING.getDataType()));
fields.add(new RecordField("type", RecordFieldType.STRING.getDataType()));
fields.add(new RecordField("dynamicPropertySource1", RecordFieldType.STRING.getDataType()));
fields.add(new RecordField("dynamicPropertySource2", RecordFieldType.STRING.getDataType()));
RecordSchema schema = new SimpleRecordSchema(fields);
Map<String, Object> valueMap = new HashMap<>();
valueMap.put("description", "This is a test comment body.");
valueMap.put("subject", "Test subject");
valueMap.put("priority", "High");
valueMap.put("type", "Development");
valueMap.put("dynamicPropertySource1", "This is a dynamic property 1");
valueMap.put("dynamicPropertySource2", "This is a dynamic property 2");
Record record = new MapRecord(schema, valueMap);
recordSet = RecordSet.of(schema, record);
}
private void initMultipleTestRecord() {
List<RecordField> fields = new ArrayList<>();
fields.add(new RecordField("description", RecordFieldType.STRING.getDataType()));
fields.add(new RecordField("priority", RecordFieldType.STRING.getDataType()));
RecordSchema schema = new SimpleRecordSchema(fields);
Map<String, Object> valueMap1 = new HashMap<>();
valueMap1.put("description", "This is a test comment body.");
valueMap1.put("priority", "High");
Record record1 = new MapRecord(schema, valueMap1);
Map<String, Object> valueMap2 = new HashMap<>();
valueMap2.put("description", "This is another test comment body.");
valueMap2.put("priority", "Low");
Record record2 = new MapRecord(schema, valueMap2);
recordSet = RecordSet.of(schema, record1, record2);
}
private void initDuplicateRecords() {
List<RecordField> fields = new ArrayList<>();
fields.add(new RecordField("description", RecordFieldType.STRING.getDataType()));
RecordSchema schema = new SimpleRecordSchema(fields);
Map<String, Object> valueMap1 = new HashMap<>();
valueMap1.put("description", "This is a test comment body.");
Record record1 = new MapRecord(schema, valueMap1);
Map<String, Object> valueMap2 = new HashMap<>();
valueMap2.put("description", "This is a test comment body.");
Record record2 = new MapRecord(schema, valueMap2);
recordSet = RecordSet.of(schema, record1, record2);
}
@Test
public void testSendMessageWithFixPropertiesAndSingleTicket() throws IOException, InterruptedException {
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_SUBJECT_NAME, "@{/subject}");
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_PRIORITY_NAME, "@{/priority}");
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_TYPE_NAME, "@{/type}");
testRunner.assertValid(sinkZendeskTicket);
testRunner.enableControllerService(sinkZendeskTicket);
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
initSingleTestRecord();
WriteResult writeResult = sinkZendeskTicket.sendData(recordSet, Collections.emptyMap(), false);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKET_RESOURCE, recordedRequest.getPath());
assertNotNull(writeResult);
assertEquals(1, writeResult.getRecordCount());
assertEquals(Collections.EMPTY_MAP, writeResult.getAttributes());
String expectedBody =
"{\n" +
" \"ticket\" : {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test comment body.\"\n" +
" },\n" +
" \"subject\" : \"Test subject\",\n" +
" \"priority\" : \"High\",\n" +
" \"type\" : \"Development\"\n" +
" }\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
}
@Test
public void testSendMessageWithFixPropertiesAndMultipleTickets() throws IOException, InterruptedException {
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_PRIORITY_NAME, "@{/priority}");
testRunner.assertValid(sinkZendeskTicket);
testRunner.enableControllerService(sinkZendeskTicket);
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
initMultipleTestRecord();
WriteResult writeResult = sinkZendeskTicket.sendData(recordSet, Collections.emptyMap(), false);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKETS_RESOURCE, recordedRequest.getPath());
assertNotNull(writeResult);
assertEquals(2, writeResult.getRecordCount());
assertEquals(Collections.EMPTY_MAP, writeResult.getAttributes());
String expectedBody =
"{\n" +
" \"tickets\" : [ {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test comment body.\"\n" +
" },\n" +
" \"priority\" : \"High\"\n" +
" }, {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is another test comment body.\"\n" +
" },\n" +
" \"priority\" : \"Low\"\n" +
" } ]\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
}
@Test
public void testSendMessageWithRecordPathDynamicProperties() throws IOException, InterruptedException {
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
testRunner.setProperty(sinkZendeskTicket, "/dp1/dynamicPropertyTarget1", "@{/dynamicPropertySource1}");
testRunner.setProperty(sinkZendeskTicket, "/dp1/dp2/dp3/dynamicPropertyTarget2", "@{/dynamicPropertySource2}");
testRunner.assertValid(sinkZendeskTicket);
testRunner.enableControllerService(sinkZendeskTicket);
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
initSingleTestRecord();
WriteResult writeResult = sinkZendeskTicket.sendData(recordSet, Collections.emptyMap(), false);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKET_RESOURCE, recordedRequest.getPath());
assertNotNull(writeResult);
assertEquals(1, writeResult.getRecordCount());
assertEquals(Collections.EMPTY_MAP, writeResult.getAttributes());
String expectedBody =
"{\n" +
" \"ticket\" : {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test comment body.\"\n" +
" },\n" +
" \"dp1\" : {\n" +
" \"dp2\" : {\n" +
" \"dp3\" : {\n" +
" \"dynamicPropertyTarget2\" : \"This is a dynamic property 2\"\n" +
" }\n" +
" },\n" +
" \"dynamicPropertyTarget1\" : \"This is a dynamic property 1\"\n" +
" }\n" +
" }\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
}
@Test
public void testSendMessageWithConstantDynamicProperties() throws IOException, InterruptedException {
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
testRunner.setProperty(sinkZendeskTicket, "/dp1/dynamicPropertyTarget1", "Constant 1");
testRunner.setProperty(sinkZendeskTicket, "/dp1/dp2/dp3/dynamicPropertyTarget2", "Constant 2");
testRunner.assertValid(sinkZendeskTicket);
testRunner.enableControllerService(sinkZendeskTicket);
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
initSingleTestRecord();
WriteResult writeResult = sinkZendeskTicket.sendData(recordSet, Collections.emptyMap(), false);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKET_RESOURCE, recordedRequest.getPath());
assertNotNull(writeResult);
assertEquals(1, writeResult.getRecordCount());
assertEquals(Collections.EMPTY_MAP, writeResult.getAttributes());
String expectedBody =
"{\n" +
" \"ticket\" : {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test comment body.\"\n" +
" },\n" +
" \"dp1\" : {\n" +
" \"dp2\" : {\n" +
" \"dp3\" : {\n" +
" \"dynamicPropertyTarget2\" : \"Constant 2\"\n" +
" }\n" +
" },\n" +
" \"dynamicPropertyTarget1\" : \"Constant 1\"\n" +
" }\n" +
" }\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
}
@Test
public void testRecordCache() throws IOException, InterruptedException {
testRunner.setProperty(sinkZendeskTicket, ZENDESK_TICKET_COMMENT_BODY_NAME, "@{/description}");
testRunner.assertValid(sinkZendeskTicket);
testRunner.enableControllerService(sinkZendeskTicket);
server.enqueue(new MockResponse().setResponseCode(HTTP_OK).setBody(EMPTY_RESPONSE));
initDuplicateRecords();
WriteResult writeResult = sinkZendeskTicket.sendData(recordSet, Collections.emptyMap(), false);
// then
RecordedRequest recordedRequest = server.takeRequest();
assertEquals(ZENDESK_CREATE_TICKET_RESOURCE, recordedRequest.getPath());
assertNotNull(writeResult);
assertEquals(1, writeResult.getRecordCount());
assertEquals(Collections.EMPTY_MAP, writeResult.getAttributes());
String expectedBody =
"{\n" +
" \"ticket\" : {\n" +
" \"comment\" : {\n" +
" \"body\" : \"This is a test comment body.\"\n" +
" }\n" +
" }\n" +
"}";
assertEquals(OBJECT_MAPPER.readTree(expectedBody), OBJECT_MAPPER.readTree(recordedRequest.getBody().inputStream()));
}
class TestZendeskRecordSink extends ZendeskRecordSink {
@Override
HttpUriBuilder uriBuilder(String resourcePath) {
HttpUrl url = server.url(resourcePath);
return new StandardHttpUriBuilder()
.scheme(url.scheme())
.host(url.host())
.port(url.port())
.encodedPath(url.encodedPath());
}
}
}

3
nifi-nar-bundles/nifi-zendesk-bundle/pom.xml
View File

@ -29,5 +29,8 @@
<modules>
<module>nifi-zendesk-processors</module>
<module>nifi-zendesk-nar</module>
<module>nifi-zendesk-common</module>
<module>nifi-zendesk-services</module>
<module>nifi-zendesk-services-nar</module>
</modules>
</project>