nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/StatelessKafkaConnectorUtil.java - nifi - Git at Google

 /*
  * 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.kafka.connect;

 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.nifi.kafka.connect.validators.ConnectDirectoryExistsValidator;
 import org.apache.nifi.kafka.connect.validators.ConnectHttpUrlValidator;
 import org.apache.nifi.kafka.connect.validators.FlowSnapshotValidator;
 import org.apache.nifi.stateless.bootstrap.StatelessBootstrap;
 import org.apache.nifi.stateless.config.ExtensionClientDefinition;
 import org.apache.nifi.stateless.config.ParameterOverride;
 import org.apache.nifi.stateless.config.SslContextDefinition;
 import org.apache.nifi.stateless.engine.StatelessEngineConfiguration;
 import org.apache.nifi.stateless.flow.DataflowDefinition;
 import org.apache.nifi.stateless.flow.StatelessDataflow;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;

 import java.io.File;
 import java.io.IOException;
 import java.net.URL;
 import java.net.URLClassLoader;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.jar.JarFile;
 import java.util.jar.Manifest;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;

 import static org.apache.kafka.common.config.ConfigDef.NonEmptyStringWithoutControlChars.nonEmptyStringWithoutControlChars;

 public class StatelessKafkaConnectorUtil {
     private static final String UNKNOWN_VERSION = "<Unable to determine Stateless NiFi Kafka Connector Version>";
     private static final Logger logger = LoggerFactory.getLogger(StatelessKafkaConnectorUtil.class);
     private static final Lock unpackNarLock = new ReentrantLock();

     static final String NAR_DIRECTORY = "nar.directory";
     static final String EXTENSIONS_DIRECTORY = "extensions.directory";
     static final String WORKING_DIRECTORY = "working.directory";
     static final String FLOW_SNAPSHOT = "flow.snapshot";
     static final String KRB5_FILE = "krb5.file";
     static final String NEXUS_BASE_URL = "nexus.url";
     static final String DATAFLOW_TIMEOUT = "dataflow.timeout";
     static final String DATAFLOW_NAME = "name";

     static final String TRUSTSTORE_FILE = "security.truststore";
     static final String TRUSTSTORE_TYPE = "security.truststoreType";
     static final String TRUSTSTORE_PASSWORD = "security.truststorePasswd";
     static final String KEYSTORE_FILE = "security.keystore";
     static final String KEYSTORE_TYPE = "security.keystoreType";
     static final String KEYSTORE_PASSWORD = "security.keystorePasswd";
     static final String KEY_PASSWORD = "security.keyPasswd";
     static final String SENSITIVE_PROPS_KEY = "sensitive.props.key";

     static final String BOOTSTRAP_SNAPSHOT_URL = "nifi.stateless.flow.snapshot.url";
     static final String BOOTSTRAP_SNAPSHOT_FILE = "nifi.stateless.flow.snapshot.file";
     static final String BOOTSTRAP_SNAPSHOT_CONTENTS = "nifi.stateless.flow.snapshot.contents";
     static final String BOOTSTRAP_FLOW_NAME = "nifi.stateless.flow.name";

     static final String DEFAULT_KRB5_FILE = "/etc/krb5.conf";
     static final String DEFAULT_DATAFLOW_TIMEOUT = "60 sec";
     static final File DEFAULT_WORKING_DIRECTORY = new File("/tmp/nifi-stateless-working");
     static final File DEFAULT_EXTENSIONS_DIRECTORY = new File("/tmp/nifi-stateless-extensions");
     static final String DEFAULT_SENSITIVE_PROPS_KEY = "nifi-stateless";

     private static final Pattern STATELESS_BOOTSTRAP_FILE_PATTERN = Pattern.compile("nifi-stateless-bootstrap-(.*).jar");
     private static final Pattern PARAMETER_WITH_CONTEXT_PATTERN = Pattern.compile("parameter\\.(.*?):(.*)");
     private static final Pattern PARAMETER_WITHOUT_CONTEXT_PATTERN = Pattern.compile("parameter\\.(.*)");

     public static void addCommonConfigElements(final ConfigDef configDef) {
         configDef.define(NAR_DIRECTORY, ConfigDef.Type.STRING, null, new ConnectDirectoryExistsValidator(), ConfigDef.Importance.HIGH,
             "Specifies the directory that stores the NiFi Archives (NARs)");
         configDef.define(EXTENSIONS_DIRECTORY, ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH,
             "Specifies the directory that stores the extensions that will be downloaded (if any) from the configured Extension Client");
         configDef.define(WORKING_DIRECTORY, ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH,
             "Specifies the temporary working directory for expanding NiFi Archives (NARs)");
         configDef.define(FLOW_SNAPSHOT, ConfigDef.Type.STRING, null, new FlowSnapshotValidator(), ConfigDef.Importance.HIGH,
             "Specifies the dataflow to run. This may be a file containing the dataflow, a URL that points to a dataflow, or a String containing the entire dataflow as an escaped JSON.");
         configDef.define(DATAFLOW_NAME, ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE, nonEmptyStringWithoutControlChars(), ConfigDef.Importance.HIGH, "The name of the dataflow.");

         configDef.define(StatelessKafkaConnectorUtil.KRB5_FILE, ConfigDef.Type.STRING, StatelessKafkaConnectorUtil.DEFAULT_KRB5_FILE, ConfigDef.Importance.MEDIUM,
             "Specifies the krb5.conf file to use if connecting to Kerberos-enabled services");
         configDef.define(StatelessKafkaConnectorUtil.NEXUS_BASE_URL, ConfigDef.Type.STRING, null, new ConnectHttpUrlValidator(), ConfigDef.Importance.MEDIUM,
             "Specifies the Base URL of the Nexus instance to source extensions from");

         configDef.define(StatelessKafkaConnectorUtil.DATAFLOW_TIMEOUT, ConfigDef.Type.STRING, StatelessKafkaConnectorUtil.DEFAULT_DATAFLOW_TIMEOUT, ConfigDef.Importance.MEDIUM,
             "Specifies the amount of time to wait for the dataflow to finish processing input before considering the dataflow a failure");

         configDef.define(KEYSTORE_FILE, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
             "Filename of the keystore that Stateless NiFi should use for connecting to NiFi Registry and for Site-to-Site communications.");
         configDef.define(KEYSTORE_TYPE, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
             "The type of the Keystore file. Either JKS or PKCS12.");
         configDef.define(KEYSTORE_PASSWORD, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.MEDIUM,
             "The password for the keystore.");
         configDef.define(KEY_PASSWORD, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.MEDIUM,
             "The password for the key in the keystore. If not provided, the password is assumed to be the same as the keystore password.");
         configDef.define(TRUSTSTORE_FILE, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
             "Filename of the truststore that Stateless NiFi should use for connecting to NiFi Registry and for Site-to-Site communications. If not specified, communications will occur only over " +
                 "http, not https.");
         configDef.define(TRUSTSTORE_TYPE, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
             "The type of the Truststore file. Either JKS or PKCS12.");
         configDef.define(TRUSTSTORE_PASSWORD, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.MEDIUM,
             "The password for the truststore.");
         configDef.define(SENSITIVE_PROPS_KEY, ConfigDef.Type.PASSWORD, DEFAULT_SENSITIVE_PROPS_KEY, ConfigDef.Importance.MEDIUM, "A key that components can use for encrypting and decrypting " +
             "sensitive values.");
     }

     public static String getVersion() {
         final File bootstrapJar = detectBootstrapJar();
         if (bootstrapJar == null) {
             return UNKNOWN_VERSION;
         }

         try (final JarFile jarFile = new JarFile(bootstrapJar)) {
             final Manifest manifest = jarFile.getManifest();
             if (manifest != null) {
                 return manifest.getMainAttributes().getValue("Implementation-Version");
             }
         } catch (IOException e) {
             logger.warn("Could not determine Version of NiFi Stateless Kafka Connector", e);
             return UNKNOWN_VERSION;
         }

         return UNKNOWN_VERSION;
     }

     public static StatelessDataflow createDataflow(final Map<String, String> properties) {
         final StatelessEngineConfiguration engineConfiguration = createEngineConfiguration(properties);
         final String configuredFlowSnapshot = properties.get(FLOW_SNAPSHOT);

         final List<ParameterOverride> parameterOverrides = parseParameterOverrides(properties);
         final String dataflowName = properties.get(DATAFLOW_NAME);

         final DataflowDefinition<?> dataflowDefinition;
         final StatelessBootstrap bootstrap;
         try {
             final Map<String, String> dataflowDefinitionProperties = new HashMap<>();

             if (configuredFlowSnapshot.startsWith("http://") || configuredFlowSnapshot.startsWith("https://")) {
                 logger.debug("Configured Flow Snapshot appears to be a URL. Will use {} property to configured Stateless NiFi", BOOTSTRAP_SNAPSHOT_URL);
                 dataflowDefinitionProperties.put(BOOTSTRAP_SNAPSHOT_URL, configuredFlowSnapshot);
             } else if (configuredFlowSnapshot.trim().startsWith("{")) {
                 logger.debug("Configured Flow Snapshot appears to be JSON. Will use {} property to configured Stateless NiFi", BOOTSTRAP_SNAPSHOT_CONTENTS);
                 dataflowDefinitionProperties.put(BOOTSTRAP_SNAPSHOT_CONTENTS, configuredFlowSnapshot);
             } else {
                 logger.debug("Configured Flow Snapshot appears to be a File. Will use {} property to configured Stateless NiFi", BOOTSTRAP_SNAPSHOT_FILE);
                 final File flowSnapshotFile = new File(configuredFlowSnapshot);
                 dataflowDefinitionProperties.put(BOOTSTRAP_SNAPSHOT_FILE, flowSnapshotFile.getAbsolutePath());
             }

             dataflowDefinitionProperties.put(BOOTSTRAP_FLOW_NAME, dataflowName);

             MDC.setContextMap(Collections.singletonMap("dataflow", dataflowName));

             // Use a Write Lock to ensure that only a single thread is calling StatelessBootstrap.bootstrap().
             // We do this because the bootstrap() method will expand all NAR files into the working directory.
             // If we have multiple Connector instances, or multiple tasks, we don't want several threads all
             // unpacking NARs at the same time, as it could potentially result in the working directory becoming corrupted.
             unpackNarLock.lock();
             try {
                 bootstrap = StatelessBootstrap.bootstrap(engineConfiguration, StatelessNiFiSourceTask.class.getClassLoader());
             } finally {
                 unpackNarLock.unlock();
             }

             dataflowDefinition = bootstrap.parseDataflowDefinition(dataflowDefinitionProperties, parameterOverrides);
             return bootstrap.createDataflow(dataflowDefinition);
         } catch (final Exception e) {
             throw new RuntimeException("Failed to bootstrap Stateless NiFi Engine", e);
         }
     }

     private static List<ParameterOverride> parseParameterOverrides(final Map<String, String> properties) {
         final List<ParameterOverride> parameterOverrides = new ArrayList<>();

         for (final Map.Entry<String, String> entry : properties.entrySet()) {
             final String parameterValue = entry.getValue();

             ParameterOverride parameterOverride = null;
             final Matcher matcher = PARAMETER_WITH_CONTEXT_PATTERN.matcher(entry.getKey());
             if (matcher.matches()) {
                 final String contextName = matcher.group(1);
                 final String parameterName = matcher.group(2);
                 parameterOverride = new ParameterOverride(contextName, parameterName, parameterValue);
             } else {
                 final Matcher noContextMatcher = PARAMETER_WITHOUT_CONTEXT_PATTERN.matcher(entry.getKey());
                 if (noContextMatcher.matches()) {
                     final String parameterName = noContextMatcher.group(1);
                     parameterOverride = new ParameterOverride(parameterName, parameterValue);
                 }
             }

             if (parameterOverride != null) {
                 parameterOverrides.add(parameterOverride);
             }
         }

         return parameterOverrides;
     }

     public static Map<String, String> getLoggableProperties(final Map<String, String> properties) {
         final Map<String, String> loggable = new HashMap<>(properties);
         loggable.keySet().removeIf(key -> key.startsWith("parameter."));
         return loggable;
     }

     private static StatelessEngineConfiguration createEngineConfiguration(final Map<String, String> properties) {
         final File narDirectory;
         final String narDirectoryFilename = properties.get(NAR_DIRECTORY);
         if (narDirectoryFilename == null) {
             narDirectory = detectNarDirectory();
         } else {
             narDirectory = new File(narDirectoryFilename);
         }

         final String dataflowName = properties.get(DATAFLOW_NAME);

         final File baseWorkingDirectory;
         final String workingDirectoryFilename = properties.get(WORKING_DIRECTORY);
         if (workingDirectoryFilename == null) {
             baseWorkingDirectory = DEFAULT_WORKING_DIRECTORY;
         } else {
             baseWorkingDirectory = new File(workingDirectoryFilename);
         }
         final File workingDirectory = new File(baseWorkingDirectory, dataflowName);

         final File extensionsDirectory;
         final String extensionsDirectoryFilename = properties.get(EXTENSIONS_DIRECTORY);
         if (extensionsDirectoryFilename == null) {
             extensionsDirectory = DEFAULT_EXTENSIONS_DIRECTORY;
         } else {
             extensionsDirectory = new File(extensionsDirectoryFilename);
         }

         final SslContextDefinition sslContextDefinition = createSslContextDefinition(properties);

         final StatelessEngineConfiguration engineConfiguration = new StatelessEngineConfiguration() {
             @Override
             public File getWorkingDirectory() {
                 return workingDirectory;
             }

             @Override
             public File getNarDirectory() {
                 return narDirectory;
             }

             @Override
             public File getExtensionsDirectory() {
                 return extensionsDirectory;
             }

             @Override
             public File getKrb5File() {
                 return new File(properties.getOrDefault(KRB5_FILE, DEFAULT_KRB5_FILE));
             }

             @Override
             public Optional<File> getContentRepositoryDirectory() {
                 return Optional.empty();
             }

             @Override
             public SslContextDefinition getSslContext() {
                 return sslContextDefinition;
             }

             @Override
             public String getSensitivePropsKey() {
                 return properties.getOrDefault(SENSITIVE_PROPS_KEY, DEFAULT_SENSITIVE_PROPS_KEY);
             }

             @Override
             public List<ExtensionClientDefinition> getExtensionClients() {
                 final List<ExtensionClientDefinition> extensionClientDefinitions = new ArrayList<>();

                 final String nexusBaseUrl = properties.get(NEXUS_BASE_URL);
                 if (nexusBaseUrl != null) {
                     final ExtensionClientDefinition definition = new ExtensionClientDefinition();
                     definition.setUseSslContext(false);
                     definition.setExtensionClientType("nexus");
                     definition.setCommsTimeout("30 secs");
                     definition.setBaseUrl(nexusBaseUrl);
                     extensionClientDefinitions.add(definition);
                 }

                 return extensionClientDefinitions;
             }
         };

         return engineConfiguration;
     }

     private static SslContextDefinition createSslContextDefinition(final Map<String, String> properties) {
         final String truststoreFile = properties.get(TRUSTSTORE_FILE);
         if (truststoreFile == null || truststoreFile.trim().isEmpty()) {
             return null;
         }

         final SslContextDefinition sslContextDefinition;
         sslContextDefinition = new SslContextDefinition();
         sslContextDefinition.setTruststoreFile(truststoreFile);
         sslContextDefinition.setTruststorePass(properties.get(TRUSTSTORE_PASSWORD));
         sslContextDefinition.setTruststoreType(properties.get(TRUSTSTORE_TYPE));

         final String keystoreFile = properties.get(KEYSTORE_FILE);
         if (keystoreFile != null && !keystoreFile.trim().isEmpty()) {
             sslContextDefinition.setKeystoreFile(keystoreFile);
             sslContextDefinition.setKeystoreType(properties.get(KEYSTORE_TYPE));

             final String keystorePass = properties.get(KEYSTORE_PASSWORD);
             sslContextDefinition.setKeystorePass(keystorePass);

             final String explicitKeyPass = properties.get(KEY_PASSWORD);
             final String keyPass = (explicitKeyPass == null || explicitKeyPass.trim().isEmpty()) ? keystorePass : explicitKeyPass;
             sslContextDefinition.setKeyPass(keyPass);
         }

         return sslContextDefinition;
     }

     private static URLClassLoader getConnectClassLoader() {
         final ClassLoader classLoader = StatelessKafkaConnectorUtil.class.getClassLoader();
         if (!(classLoader instanceof URLClassLoader)) {
             throw new IllegalStateException("No configuration value was set for the " + NAR_DIRECTORY + " configuration property, and was unable to determine the NAR directory automatically");
         }

         return (URLClassLoader) classLoader;
     }

     private static File detectBootstrapJar() {
         final URLClassLoader urlClassLoader = getConnectClassLoader();
         for (final URL url : urlClassLoader.getURLs()) {
             final String artifactFilename = url.getFile();
             if (artifactFilename == null) {
                 continue;
             }

             final File artifactFile = new File(artifactFilename);
             if (STATELESS_BOOTSTRAP_FILE_PATTERN.matcher(artifactFile.getName()).matches()) {
                 return artifactFile;
             }
         }

         return null;
     }

     private static File detectNarDirectory() {
         final File bootstrapJar = detectBootstrapJar();
         if (bootstrapJar == null) {
             final URLClassLoader urlClassLoader = getConnectClassLoader();
             logger.error("ClassLoader that loaded Stateless Kafka Connector did not contain nifi-stateless-bootstrap. URLs that were present: {}", Arrays.asList(urlClassLoader.getURLs()));
             throw new IllegalStateException("No configuration value was set for the " + NAR_DIRECTORY + " configuration property, and was unable to determine the NAR directory automatically");
         }

         final File narDirectory = bootstrapJar.getParentFile();
         logger.info("Detected NAR Directory to be {}", narDirectory.getAbsolutePath());
         return narDirectory;
     }
 }
	/*
	* 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.kafka.connect;

	import org.apache.kafka.common.config.ConfigDef;
	import org.apache.nifi.kafka.connect.validators.ConnectDirectoryExistsValidator;
	import org.apache.nifi.kafka.connect.validators.ConnectHttpUrlValidator;
	import org.apache.nifi.kafka.connect.validators.FlowSnapshotValidator;
	import org.apache.nifi.stateless.bootstrap.StatelessBootstrap;
	import org.apache.nifi.stateless.config.ExtensionClientDefinition;
	import org.apache.nifi.stateless.config.ParameterOverride;
	import org.apache.nifi.stateless.config.SslContextDefinition;
	import org.apache.nifi.stateless.engine.StatelessEngineConfiguration;
	import org.apache.nifi.stateless.flow.DataflowDefinition;
	import org.apache.nifi.stateless.flow.StatelessDataflow;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.slf4j.MDC;

	import java.io.File;
	import java.io.IOException;
	import java.net.URL;
	import java.net.URLClassLoader;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.Optional;
	import java.util.concurrent.locks.Lock;
	import java.util.concurrent.locks.ReentrantLock;
	import java.util.jar.JarFile;
	import java.util.jar.Manifest;
	import java.util.regex.Matcher;
	import java.util.regex.Pattern;

	import static org.apache.kafka.common.config.ConfigDef.NonEmptyStringWithoutControlChars.nonEmptyStringWithoutControlChars;

	public class StatelessKafkaConnectorUtil {
	private static final String UNKNOWN_VERSION = "<Unable to determine Stateless NiFi Kafka Connector Version>";
	private static final Logger logger = LoggerFactory.getLogger(StatelessKafkaConnectorUtil.class);
	private static final Lock unpackNarLock = new ReentrantLock();

	static final String NAR_DIRECTORY = "nar.directory";
	static final String EXTENSIONS_DIRECTORY = "extensions.directory";
	static final String WORKING_DIRECTORY = "working.directory";
	static final String FLOW_SNAPSHOT = "flow.snapshot";
	static final String KRB5_FILE = "krb5.file";
	static final String NEXUS_BASE_URL = "nexus.url";
	static final String DATAFLOW_TIMEOUT = "dataflow.timeout";
	static final String DATAFLOW_NAME = "name";

	static final String TRUSTSTORE_FILE = "security.truststore";
	static final String TRUSTSTORE_TYPE = "security.truststoreType";
	static final String TRUSTSTORE_PASSWORD = "security.truststorePasswd";
	static final String KEYSTORE_FILE = "security.keystore";
	static final String KEYSTORE_TYPE = "security.keystoreType";
	static final String KEYSTORE_PASSWORD = "security.keystorePasswd";
	static final String KEY_PASSWORD = "security.keyPasswd";
	static final String SENSITIVE_PROPS_KEY = "sensitive.props.key";

	static final String BOOTSTRAP_SNAPSHOT_URL = "nifi.stateless.flow.snapshot.url";
	static final String BOOTSTRAP_SNAPSHOT_FILE = "nifi.stateless.flow.snapshot.file";
	static final String BOOTSTRAP_SNAPSHOT_CONTENTS = "nifi.stateless.flow.snapshot.contents";
	static final String BOOTSTRAP_FLOW_NAME = "nifi.stateless.flow.name";

	static final String DEFAULT_KRB5_FILE = "/etc/krb5.conf";
	static final String DEFAULT_DATAFLOW_TIMEOUT = "60 sec";
	static final File DEFAULT_WORKING_DIRECTORY = new File("/tmp/nifi-stateless-working");
	static final File DEFAULT_EXTENSIONS_DIRECTORY = new File("/tmp/nifi-stateless-extensions");
	static final String DEFAULT_SENSITIVE_PROPS_KEY = "nifi-stateless";

	private static final Pattern STATELESS_BOOTSTRAP_FILE_PATTERN = Pattern.compile("nifi-stateless-bootstrap-(.*).jar");
	private static final Pattern PARAMETER_WITH_CONTEXT_PATTERN = Pattern.compile("parameter\\.(.?):(.)");
	private static final Pattern PARAMETER_WITHOUT_CONTEXT_PATTERN = Pattern.compile("parameter\\.(.*)");

	public static void addCommonConfigElements(final ConfigDef configDef) {
	configDef.define(NAR_DIRECTORY, ConfigDef.Type.STRING, null, new ConnectDirectoryExistsValidator(), ConfigDef.Importance.HIGH,
	"Specifies the directory that stores the NiFi Archives (NARs)");
	configDef.define(EXTENSIONS_DIRECTORY, ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH,
	"Specifies the directory that stores the extensions that will be downloaded (if any) from the configured Extension Client");
	configDef.define(WORKING_DIRECTORY, ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH,
	"Specifies the temporary working directory for expanding NiFi Archives (NARs)");
	configDef.define(FLOW_SNAPSHOT, ConfigDef.Type.STRING, null, new FlowSnapshotValidator(), ConfigDef.Importance.HIGH,
	"Specifies the dataflow to run. This may be a file containing the dataflow, a URL that points to a dataflow, or a String containing the entire dataflow as an escaped JSON.");
	configDef.define(DATAFLOW_NAME, ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE, nonEmptyStringWithoutControlChars(), ConfigDef.Importance.HIGH, "The name of the dataflow.");

	configDef.define(StatelessKafkaConnectorUtil.KRB5_FILE, ConfigDef.Type.STRING, StatelessKafkaConnectorUtil.DEFAULT_KRB5_FILE, ConfigDef.Importance.MEDIUM,
	"Specifies the krb5.conf file to use if connecting to Kerberos-enabled services");
	configDef.define(StatelessKafkaConnectorUtil.NEXUS_BASE_URL, ConfigDef.Type.STRING, null, new ConnectHttpUrlValidator(), ConfigDef.Importance.MEDIUM,
	"Specifies the Base URL of the Nexus instance to source extensions from");

	configDef.define(StatelessKafkaConnectorUtil.DATAFLOW_TIMEOUT, ConfigDef.Type.STRING, StatelessKafkaConnectorUtil.DEFAULT_DATAFLOW_TIMEOUT, ConfigDef.Importance.MEDIUM,
	"Specifies the amount of time to wait for the dataflow to finish processing input before considering the dataflow a failure");

	configDef.define(KEYSTORE_FILE, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
	"Filename of the keystore that Stateless NiFi should use for connecting to NiFi Registry and for Site-to-Site communications.");
	configDef.define(KEYSTORE_TYPE, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
	"The type of the Keystore file. Either JKS or PKCS12.");
	configDef.define(KEYSTORE_PASSWORD, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.MEDIUM,
	"The password for the keystore.");
	configDef.define(KEY_PASSWORD, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.MEDIUM,
	"The password for the key in the keystore. If not provided, the password is assumed to be the same as the keystore password.");
	configDef.define(TRUSTSTORE_FILE, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
	"Filename of the truststore that Stateless NiFi should use for connecting to NiFi Registry and for Site-to-Site communications. If not specified, communications will occur only over " +
	"http, not https.");
	configDef.define(TRUSTSTORE_TYPE, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
	"The type of the Truststore file. Either JKS or PKCS12.");
	configDef.define(TRUSTSTORE_PASSWORD, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.MEDIUM,
	"The password for the truststore.");
	configDef.define(SENSITIVE_PROPS_KEY, ConfigDef.Type.PASSWORD, DEFAULT_SENSITIVE_PROPS_KEY, ConfigDef.Importance.MEDIUM, "A key that components can use for encrypting and decrypting " +
	"sensitive values.");
	}

	public static String getVersion() {
	final File bootstrapJar = detectBootstrapJar();
	if (bootstrapJar == null) {
	return UNKNOWN_VERSION;
	}

	try (final JarFile jarFile = new JarFile(bootstrapJar)) {
	final Manifest manifest = jarFile.getManifest();
	if (manifest != null) {
	return manifest.getMainAttributes().getValue("Implementation-Version");
	}
	} catch (IOException e) {
	logger.warn("Could not determine Version of NiFi Stateless Kafka Connector", e);
	return UNKNOWN_VERSION;
	}

	return UNKNOWN_VERSION;
	}

	public static StatelessDataflow createDataflow(final Map<String, String> properties) {
	final StatelessEngineConfiguration engineConfiguration = createEngineConfiguration(properties);
	final String configuredFlowSnapshot = properties.get(FLOW_SNAPSHOT);

	final List<ParameterOverride> parameterOverrides = parseParameterOverrides(properties);
	final String dataflowName = properties.get(DATAFLOW_NAME);

	final DataflowDefinition<?> dataflowDefinition;
	final StatelessBootstrap bootstrap;
	try {
	final Map<String, String> dataflowDefinitionProperties = new HashMap<>();

	if (configuredFlowSnapshot.startsWith("http://") \|\| configuredFlowSnapshot.startsWith("https://")) {
	logger.debug("Configured Flow Snapshot appears to be a URL. Will use {} property to configured Stateless NiFi", BOOTSTRAP_SNAPSHOT_URL);
	dataflowDefinitionProperties.put(BOOTSTRAP_SNAPSHOT_URL, configuredFlowSnapshot);
	} else if (configuredFlowSnapshot.trim().startsWith("{")) {
	logger.debug("Configured Flow Snapshot appears to be JSON. Will use {} property to configured Stateless NiFi", BOOTSTRAP_SNAPSHOT_CONTENTS);
	dataflowDefinitionProperties.put(BOOTSTRAP_SNAPSHOT_CONTENTS, configuredFlowSnapshot);
	} else {
	logger.debug("Configured Flow Snapshot appears to be a File. Will use {} property to configured Stateless NiFi", BOOTSTRAP_SNAPSHOT_FILE);
	final File flowSnapshotFile = new File(configuredFlowSnapshot);
	dataflowDefinitionProperties.put(BOOTSTRAP_SNAPSHOT_FILE, flowSnapshotFile.getAbsolutePath());
	}

	dataflowDefinitionProperties.put(BOOTSTRAP_FLOW_NAME, dataflowName);

	MDC.setContextMap(Collections.singletonMap("dataflow", dataflowName));

	// Use a Write Lock to ensure that only a single thread is calling StatelessBootstrap.bootstrap().
	// We do this because the bootstrap() method will expand all NAR files into the working directory.
	// If we have multiple Connector instances, or multiple tasks, we don't want several threads all
	// unpacking NARs at the same time, as it could potentially result in the working directory becoming corrupted.
	unpackNarLock.lock();
	try {
	bootstrap = StatelessBootstrap.bootstrap(engineConfiguration, StatelessNiFiSourceTask.class.getClassLoader());
	} finally {
	unpackNarLock.unlock();
	}

	dataflowDefinition = bootstrap.parseDataflowDefinition(dataflowDefinitionProperties, parameterOverrides);
	return bootstrap.createDataflow(dataflowDefinition);
	} catch (final Exception e) {
	throw new RuntimeException("Failed to bootstrap Stateless NiFi Engine", e);
	}
	}

	private static List<ParameterOverride> parseParameterOverrides(final Map<String, String> properties) {
	final List<ParameterOverride> parameterOverrides = new ArrayList<>();

	for (final Map.Entry<String, String> entry : properties.entrySet()) {
	final String parameterValue = entry.getValue();

	ParameterOverride parameterOverride = null;
	final Matcher matcher = PARAMETER_WITH_CONTEXT_PATTERN.matcher(entry.getKey());
	if (matcher.matches()) {
	final String contextName = matcher.group(1);
	final String parameterName = matcher.group(2);
	parameterOverride = new ParameterOverride(contextName, parameterName, parameterValue);
	} else {
	final Matcher noContextMatcher = PARAMETER_WITHOUT_CONTEXT_PATTERN.matcher(entry.getKey());
	if (noContextMatcher.matches()) {
	final String parameterName = noContextMatcher.group(1);
	parameterOverride = new ParameterOverride(parameterName, parameterValue);
	}
	}

	if (parameterOverride != null) {
	parameterOverrides.add(parameterOverride);
	}
	}

	return parameterOverrides;
	}

	public static Map<String, String> getLoggableProperties(final Map<String, String> properties) {
	final Map<String, String> loggable = new HashMap<>(properties);
	loggable.keySet().removeIf(key -> key.startsWith("parameter."));
	return loggable;
	}

	private static StatelessEngineConfiguration createEngineConfiguration(final Map<String, String> properties) {
	final File narDirectory;
	final String narDirectoryFilename = properties.get(NAR_DIRECTORY);
	if (narDirectoryFilename == null) {
	narDirectory = detectNarDirectory();
	} else {
	narDirectory = new File(narDirectoryFilename);
	}

	final String dataflowName = properties.get(DATAFLOW_NAME);

	final File baseWorkingDirectory;
	final String workingDirectoryFilename = properties.get(WORKING_DIRECTORY);
	if (workingDirectoryFilename == null) {
	baseWorkingDirectory = DEFAULT_WORKING_DIRECTORY;
	} else {
	baseWorkingDirectory = new File(workingDirectoryFilename);
	}
	final File workingDirectory = new File(baseWorkingDirectory, dataflowName);

	final File extensionsDirectory;
	final String extensionsDirectoryFilename = properties.get(EXTENSIONS_DIRECTORY);
	if (extensionsDirectoryFilename == null) {
	extensionsDirectory = DEFAULT_EXTENSIONS_DIRECTORY;
	} else {
	extensionsDirectory = new File(extensionsDirectoryFilename);
	}

	final SslContextDefinition sslContextDefinition = createSslContextDefinition(properties);

	final StatelessEngineConfiguration engineConfiguration = new StatelessEngineConfiguration() {
	@Override
	public File getWorkingDirectory() {
	return workingDirectory;
	}

	@Override
	public File getNarDirectory() {
	return narDirectory;
	}

	@Override
	public File getExtensionsDirectory() {
	return extensionsDirectory;
	}

	@Override
	public File getKrb5File() {
	return new File(properties.getOrDefault(KRB5_FILE, DEFAULT_KRB5_FILE));
	}

	@Override
	public Optional<File> getContentRepositoryDirectory() {
	return Optional.empty();
	}

	@Override
	public SslContextDefinition getSslContext() {
	return sslContextDefinition;
	}

	@Override
	public String getSensitivePropsKey() {
	return properties.getOrDefault(SENSITIVE_PROPS_KEY, DEFAULT_SENSITIVE_PROPS_KEY);
	}

	@Override
	public List<ExtensionClientDefinition> getExtensionClients() {
	final List<ExtensionClientDefinition> extensionClientDefinitions = new ArrayList<>();

	final String nexusBaseUrl = properties.get(NEXUS_BASE_URL);
	if (nexusBaseUrl != null) {
	final ExtensionClientDefinition definition = new ExtensionClientDefinition();
	definition.setUseSslContext(false);
	definition.setExtensionClientType("nexus");
	definition.setCommsTimeout("30 secs");
	definition.setBaseUrl(nexusBaseUrl);
	extensionClientDefinitions.add(definition);
	}

	return extensionClientDefinitions;
	}
	};

	return engineConfiguration;
	}

	private static SslContextDefinition createSslContextDefinition(final Map<String, String> properties) {
	final String truststoreFile = properties.get(TRUSTSTORE_FILE);
	if (truststoreFile == null \|\| truststoreFile.trim().isEmpty()) {
	return null;
	}

	final SslContextDefinition sslContextDefinition;
	sslContextDefinition = new SslContextDefinition();
	sslContextDefinition.setTruststoreFile(truststoreFile);
	sslContextDefinition.setTruststorePass(properties.get(TRUSTSTORE_PASSWORD));
	sslContextDefinition.setTruststoreType(properties.get(TRUSTSTORE_TYPE));

	final String keystoreFile = properties.get(KEYSTORE_FILE);
	if (keystoreFile != null && !keystoreFile.trim().isEmpty()) {
	sslContextDefinition.setKeystoreFile(keystoreFile);
	sslContextDefinition.setKeystoreType(properties.get(KEYSTORE_TYPE));

	final String keystorePass = properties.get(KEYSTORE_PASSWORD);
	sslContextDefinition.setKeystorePass(keystorePass);

	final String explicitKeyPass = properties.get(KEY_PASSWORD);
	final String keyPass = (explicitKeyPass == null \|\| explicitKeyPass.trim().isEmpty()) ? keystorePass : explicitKeyPass;
	sslContextDefinition.setKeyPass(keyPass);
	}

	return sslContextDefinition;
	}

	private static URLClassLoader getConnectClassLoader() {
	final ClassLoader classLoader = StatelessKafkaConnectorUtil.class.getClassLoader();
	if (!(classLoader instanceof URLClassLoader)) {
	throw new IllegalStateException("No configuration value was set for the " + NAR_DIRECTORY + " configuration property, and was unable to determine the NAR directory automatically");
	}

	return (URLClassLoader) classLoader;
	}

	private static File detectBootstrapJar() {
	final URLClassLoader urlClassLoader = getConnectClassLoader();
	for (final URL url : urlClassLoader.getURLs()) {
	final String artifactFilename = url.getFile();
	if (artifactFilename == null) {
	continue;
	}

	final File artifactFile = new File(artifactFilename);
	if (STATELESS_BOOTSTRAP_FILE_PATTERN.matcher(artifactFile.getName()).matches()) {
	return artifactFile;
	}
	}

	return null;
	}

	private static File detectNarDirectory() {
	final File bootstrapJar = detectBootstrapJar();
	if (bootstrapJar == null) {
	final URLClassLoader urlClassLoader = getConnectClassLoader();
	logger.error("ClassLoader that loaded Stateless Kafka Connector did not contain nifi-stateless-bootstrap. URLs that were present: {}", Arrays.asList(urlClassLoader.getURLs()));
	throw new IllegalStateException("No configuration value was set for the " + NAR_DIRECTORY + " configuration property, and was unable to determine the NAR directory automatically");
	}

	final File narDirectory = bootstrapJar.getParentFile();
	logger.info("Detected NAR Directory to be {}", narDirectory.getAbsolutePath());
	return narDirectory;
	}
	}