You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2019/10/15 08:35:15 UTC
[camel] branch master updated: CAMEL-14067 - camel-hdfs - Expose
HighAvailability configuration (ConfiguredFailoverProxyProvider) (#3248)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 920a57f CAMEL-14067 - camel-hdfs - Expose HighAvailability configuration (ConfiguredFailoverProxyProvider) (#3248)
920a57f is described below
commit 920a57f5083953d9223140a4c81d49d1dade11dd
Author: Marius Cornescu <ma...@yahoo.com>
AuthorDate: Tue Oct 15 10:34:55 2019 +0200
CAMEL-14067 - camel-hdfs - Expose HighAvailability configuration (ConfiguredFailoverProxyProvider) (#3248)
* Small fix + rename kerberosNamedNodes to namedNodes.
* refactor to use HA config regardless of kerberos.
* Refactor
* CAMEL-14067 - camel-hdfs - Expose HighAvailability configuration (ConfiguredFailoverProxyProvider)
* CAMEL-14067 - camel-hdfs - Expose HighAvailability configuration (ConfiguredFailoverProxyProvider)
---
.../camel-hdfs/src/main/docs/hdfs-component.adoc | 2 +-
.../component/hdfs/HaConfigurationBuilder.java | 79 ++++++++++++
.../apache/camel/component/hdfs/HdfsComponent.java | 6 +-
.../camel/component/hdfs/HdfsConfiguration.java | 33 +++--
.../org/apache/camel/component/hdfs/HdfsInfo.java | 23 ++--
.../hdfs/kerberos/KerberosAuthentication.java | 61 +++++++++
.../hdfs/kerberos/KerberosConfiguration.java | 138 ---------------------
.../kerberos/KerberosConfigurationBuilder.java | 67 ++++++++++
.../component/hdfs/HaConfigurationBuilderTest.java | 53 ++++++++
.../hdfs/kerberos/KerberosAuthenticationTest.java | 68 ++++++++++
.../kerberos/KerberosConfigurationBuilderTest.java | 59 +++++++++
.../hdfs/kerberos/KerberosConfigurationTest.java | 94 --------------
.../endpoint/dsl/HdfsEndpointBuilderFactory.java | 77 ++++++------
13 files changed, 457 insertions(+), 303 deletions(-)
diff --git a/components/camel-hdfs/src/main/docs/hdfs-component.adoc b/components/camel-hdfs/src/main/docs/hdfs-component.adoc
index 15213da..423f915 100644
--- a/components/camel-hdfs/src/main/docs/hdfs-component.adoc
+++ b/components/camel-hdfs/src/main/docs/hdfs-component.adoc
@@ -104,9 +104,9 @@ with the following path and query parameters:
| *fileType* (common) | The file type to use. For more details see Hadoop HDFS documentation about the various files types. | NORMAL_FILE | HdfsFileType
| *kerberosConfigFileLocation* (common) | The location of the kerb5.conf file (\https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html) | | String
| *kerberosKeytabLocation* (common) | The location of the keytab file used to authenticate with the kerberos nodes (contains pairs of kerberos principals and encrypted keys (which are derived from the Kerberos password)) | | String
-| *kerberosNamedNodes* (common) | A comma separated list of kerberos nodes (e.g. srv11.example.com:8021,srv12.example.com:8021) - see kerb5.conf file (\https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html) | | String
| *kerberosUsername* (common) | The username used to authenticate with the kerberos nodes | | String
| *keyType* (common) | The type for the key in case of sequence or map files. | NULL | WritableType
+| *namedNodes* (common) | A comma separated list of named nodes (e.g. srv11.example.com:8020,srv12.example.com:8020) | | String
| *owner* (common) | The file owner must match this owner for the consumer to pickup the file. Otherwise the file is skipped. | | String
| *valueType* (common) | The type for the key in case of sequence or map files | BYTES | WritableType
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HaConfigurationBuilder.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HaConfigurationBuilder.java
new file mode 100644
index 0000000..1d432b3
--- /dev/null
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HaConfigurationBuilder.java
@@ -0,0 +1,79 @@
+/*
+ * 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.camel.component.hdfs;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider;
+
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_CLIENT_FAILOVER_PROXY_PROVIDER_KEY_PREFIX;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_NAMENODES_KEY_PREFIX;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RPC_ADDRESS_KEY;
+
+final class HaConfigurationBuilder {
+
+ private static final String HFDS_NAMED_SERVICE = "hfdsNamedService";
+ private static final String HFDS_FS = "fs.defaultFS";
+
+ private HaConfigurationBuilder() {
+ // hidden
+ }
+
+ /**
+ * Generates the correct HA configuration (normally read from xml) based on the namedNodes:
+ * All named nodes have to be qualified: configuration.set("dfs.ha.namenodes.hfdsNamedService","namenode1,namenode2");
+ * For each named node the following entries is added
+ * <p>
+ * configuration.set("dfs.namenode.rpc-address.hfdsNamedService.namenode1", "namenode1:1234");
+ * <p>
+ * Finally the proxy provider has to be specified:
+ * <p>
+ * configuration.set("dfs.client.failover.proxy.provider.hfdsNamedService", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
+ * <p>
+ *
+ * @param namedNodes - All named nodes from the hadoop cluster
+ * @param replicationFactor - dfs replication factor
+ */
+ static void withClusterConfiguration(Configuration configuration, List<String> namedNodes, int replicationFactor) {
+ configuration.set(DFSConfigKeys.DFS_REPLICATION_KEY, Integer.toString(replicationFactor));
+ configuration.set(DFSConfigKeys.DFS_NAMESERVICES, HFDS_NAMED_SERVICE);
+ configuration.set(
+ DFSUtil.addKeySuffixes(DFS_HA_NAMENODES_KEY_PREFIX, HFDS_NAMED_SERVICE),
+ nodeToString(namedNodes.stream().map(HaConfigurationBuilder::nodeToString).collect(Collectors.joining(",")))
+ );
+
+ namedNodes.forEach(nodeName ->
+ configuration.set(
+ DFSUtil.addKeySuffixes(DFS_NAMENODE_RPC_ADDRESS_KEY, HFDS_NAMED_SERVICE, nodeToString(nodeName)),
+ nodeName)
+ );
+
+ configuration.set(DFS_CLIENT_FAILOVER_PROXY_PROVIDER_KEY_PREFIX + "." + HFDS_NAMED_SERVICE, ConfiguredFailoverProxyProvider.class.getName());
+
+ configuration.set(HFDS_FS, "hdfs://" + HFDS_NAMED_SERVICE);
+
+ }
+
+ private static String nodeToString(String nodeName) {
+ return nodeName.replaceAll(":[0-9]*", "").replaceAll("\\.", "_");
+ }
+
+}
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsComponent.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsComponent.java
index ebeafad..78c5d7c 100644
--- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsComponent.java
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsComponent.java
@@ -22,7 +22,7 @@ import java.util.Map;
import javax.security.auth.login.Configuration;
import org.apache.camel.Endpoint;
-import org.apache.camel.component.hdfs.kerberos.KerberosConfiguration;
+import org.apache.camel.component.hdfs.kerberos.KerberosConfigurationBuilder;
import org.apache.camel.spi.annotations.Component;
import org.apache.camel.support.DefaultComponent;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
@@ -34,8 +34,6 @@ public class HdfsComponent extends DefaultComponent {
private static final Logger LOG = LoggerFactory.getLogger(HdfsComponent.class);
- private static final String KERBEROS_5_SYS_ENV = "java.security.krb5.conf";
-
public HdfsComponent() {
initHdfs();
}
@@ -90,7 +88,7 @@ public class HdfsComponent extends DefaultComponent {
* @param kerberosConfigFileLocation - kerb5.conf file (https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html)
*/
public static void setKerberosConfigFile(String kerberosConfigFileLocation) {
- KerberosConfiguration.setKerberosConfigFile(kerberosConfigFileLocation);
+ KerberosConfigurationBuilder.setKerberosConfigFile(kerberosConfigFileLocation);
}
}
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConfiguration.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConfiguration.java
index 51e4884..be5bdb3 100644
--- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConfiguration.java
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConfiguration.java
@@ -87,8 +87,8 @@ public class HdfsConfiguration {
private String owner;
@UriParam
- private String kerberosNamedNodes;
- private List<String> kerberosNamedNodeList;
+ private String namedNodes;
+ private List<String> namedNodeList = Collections.emptyList();
@UriParam
private String kerberosConfigFileLocation;
@@ -188,7 +188,7 @@ public class HdfsConfiguration {
private List<HdfsProducer.SplitStrategy> getSplitStrategies(Map<String, Object> hdfsSettings) {
List<HdfsProducer.SplitStrategy> strategies = new ArrayList<>();
- splitStrategy = getString(hdfsSettings, "splitStrategy", kerberosNamedNodes);
+ splitStrategy = getString(hdfsSettings, "splitStrategy", splitStrategy);
if (isNotEmpty(splitStrategy)) {
String[] strategyElements = splitStrategy.split(",");
@@ -205,11 +205,11 @@ public class HdfsConfiguration {
return strategies;
}
- private List<String> getKerberosNamedNodeList(Map<String, Object> hdfsSettings) {
- kerberosNamedNodes = getString(hdfsSettings, "kerberosNamedNodes", kerberosNamedNodes);
+ private List<String> getNamedNodeList(Map<String, Object> hdfsSettings) {
+ namedNodes = getString(hdfsSettings, "namedNodes", namedNodes);
- if (isNotEmpty(kerberosNamedNodes)) {
- return Arrays.stream(kerberosNamedNodes.split(",")).distinct().collect(Collectors.toList());
+ if (isNotEmpty(namedNodes)) {
+ return Arrays.stream(namedNodes.split(",")).distinct().collect(Collectors.toList());
}
return Collections.emptyList();
@@ -261,7 +261,7 @@ public class HdfsConfiguration {
chunkSize = getInteger(hdfsSettings, "chunkSize", chunkSize);
splitStrategies = getSplitStrategies(hdfsSettings);
- kerberosNamedNodeList = getKerberosNamedNodeList(hdfsSettings);
+ namedNodeList = getNamedNodeList(hdfsSettings);
kerberosConfigFileLocation = getString(hdfsSettings, "kerberosConfigFileLocation", kerberosConfigFileLocation);
kerberosUsername = getString(hdfsSettings, "kerberosUsername", kerberosUsername);
kerberosKeytabLocation = getString(hdfsSettings, "kerberosKeytabLocation", kerberosKeytabLocation);
@@ -545,20 +545,19 @@ public class HdfsConfiguration {
this.owner = owner;
}
- public String getKerberosNamedNodes() {
- return kerberosNamedNodes;
+ public String getNamedNodes() {
+ return namedNodes;
}
/**
- * A comma separated list of kerberos nodes
- * (e.g. srv11.example.com:8021,srv12.example.com:8021) - see kerb5.conf file (https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html)
+ * A comma separated list of named nodes (e.g. srv11.example.com:8020,srv12.example.com:8020)
*/
- public void setKerberosNamedNodes(String kerberosNamedNodes) {
- this.kerberosNamedNodes = kerberosNamedNodes;
+ public void setNamedNodes(String namedNodes) {
+ this.namedNodes = namedNodes;
}
- public List<String> getKerberosNamedNodeList() {
- return kerberosNamedNodeList;
+ public List<String> getNamedNodeList() {
+ return namedNodeList;
}
public String getKerberosConfigFileLocation() {
@@ -596,7 +595,7 @@ public class HdfsConfiguration {
}
public boolean isKerberosAuthentication() {
- return isNotEmpty(kerberosNamedNodes) && isNotEmpty(kerberosConfigFileLocation) && isNotEmpty(kerberosUsername) && isNotEmpty(kerberosKeytabLocation);
+ return isNotEmpty(namedNodes) && isNotEmpty(kerberosConfigFileLocation) && isNotEmpty(kerberosUsername) && isNotEmpty(kerberosKeytabLocation);
}
}
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInfo.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInfo.java
index 2d96101..bd9f245 100644
--- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInfo.java
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsInfo.java
@@ -20,7 +20,8 @@ import java.io.IOException;
import java.net.URI;
import java.util.List;
-import org.apache.camel.component.hdfs.kerberos.KerberosConfiguration;
+import org.apache.camel.component.hdfs.kerberos.KerberosAuthentication;
+import org.apache.camel.component.hdfs.kerberos.KerberosConfigurationBuilder;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -49,27 +50,33 @@ public final class HdfsInfo {
return path;
}
- private Configuration newConfiguration(HdfsConfiguration endpointConfig) throws IOException {
+ private static Configuration newConfiguration(HdfsConfiguration endpointConfig) {
+ Configuration configuration = new Configuration();
+
if (endpointConfig.isKerberosAuthentication()) {
- List<String> namedNodes = endpointConfig.getKerberosNamedNodeList();
String kerberosConfigFileLocation = endpointConfig.getKerberosConfigFileLocation();
- return new KerberosConfiguration(namedNodes, kerberosConfigFileLocation, endpointConfig.getReplication());
+ KerberosConfigurationBuilder.withKerberosConfiguration(configuration, kerberosConfigFileLocation);
+
+ }
- } else {
- return new Configuration();
+ List<String> namedNodes = endpointConfig.getNamedNodeList();
+ if (!namedNodes.isEmpty()) {
+ HaConfigurationBuilder.withClusterConfiguration(configuration, endpointConfig.getNamedNodeList(), endpointConfig.getReplication());
}
+
+ return configuration;
}
/**
* this will connect to the hadoop hdfs file system, and in case of no connection
* then the hardcoded timeout in hadoop is 45 x 20 sec = 15 minutes
*/
- private FileSystem newFileSystem(Configuration configuration, String hdfsPath, HdfsConfiguration endpointConfig) throws IOException {
+ private static FileSystem newFileSystem(Configuration configuration, String hdfsPath, HdfsConfiguration endpointConfig) throws IOException {
if (endpointConfig.isKerberosAuthentication()) {
String userName = endpointConfig.getKerberosUsername();
String keytabLocation = endpointConfig.getKerberosKeytabLocation();
- ((KerberosConfiguration)configuration).loginWithKeytab(userName, keytabLocation);
+ new KerberosAuthentication(configuration, userName, keytabLocation).loginWithKeytab();
}
return FileSystem.get(URI.create(hdfsPath), configuration);
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosAuthentication.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosAuthentication.java
new file mode 100644
index 0000000..bb9ccab
--- /dev/null
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosAuthentication.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.hdfs.kerberos;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+import static java.lang.String.format;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+
+public class KerberosAuthentication {
+
+ private final String username;
+ private final String keyTabFileLocation;
+ private final Configuration configuration;
+
+ /**
+ * @param configuration - hdfs configuration
+ * @param username - Principal used to authenticate to the kerberos server
+ * @param keyTabFileLocation - keyTab file location
+ */
+ public KerberosAuthentication(Configuration configuration, String username, String keyTabFileLocation) {
+ this.configuration = configuration;
+ this.username = username;
+ this.keyTabFileLocation = keyTabFileLocation;
+ }
+
+ /**
+ * In order to connect to a hadoop cluster using Kerberos you need to add your own filesystem to the cache of the FileSystem component.
+ * This is done by setting the uri that you use in your camel route as the URI that is used to setup the connection.
+ * The URI is used as key when adding it to the cache (default functionality of the static FileSystem.get(URI, Configuration) method).
+ *
+ * @throws IOException - In case of error
+ */
+ public void loginWithKeytab() throws IOException {
+ if (!new File(keyTabFileLocation).exists()) {
+ throw new FileNotFoundException(format("KeyTab file [%s] could not be found.", keyTabFileLocation));
+ }
+ // we need to log in otherwise you cannot connect to the filesystem later on
+ UserGroupInformation.setConfiguration(configuration);
+ UserGroupInformation.loginUserFromKeytab(username, keyTabFileLocation);
+ }
+
+}
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosConfiguration.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosConfiguration.java
deleted file mode 100644
index 4422a08..0000000
--- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosConfiguration.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.camel.component.hdfs.kerberos;
-
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.List;
-import java.util.stream.Collectors;
-
-import static java.lang.String.format;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.apache.hadoop.hdfs.DFSUtil;
-import org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_CLIENT_FAILOVER_PROXY_PROVIDER_KEY_PREFIX;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_NAMENODES_KEY_PREFIX;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_RPC_ADDRESS_KEY;
-
-public class KerberosConfiguration extends Configuration {
-
- private static final Logger LOG = LoggerFactory.getLogger(KerberosConfiguration.class);
-
- private static final String HFDS_NAMED_SERVICE = "hfdsNamedService";
-
- private static final String KERBEROS_5_SYS_ENV = "java.security.krb5.conf";
-
- private static final String AUTHENTICATION_MODE = "hadoop.security.authentication";
- private static final String HFDS_FS = "fs.defaultFS";
-
- /**
- * Add all the kerberos specific settings needed for this authentication mode
- * Generates the correct HA configuration (normally read from xml) based on the namedNodes:
- * All named nodes have to be qualified: configuration.set("dfs.ha.namenodes.hfdsNamedService","namenode1,namenode2");
- * For each named node the following entries is added
- * <p>
- * configuration.set("dfs.namenode.rpc-address.hfdsNamedService.namenode1", "namenode1:1234");
- * <p>
- * Finally the proxy provider has to be specified:
- * <p>
- * configuration.set("dfs.client.failover.proxy.provider.hfdsNamedService", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
- * <p>
- *
- * @param namedNodes - All named nodes from the hadoop cluster
- * @param kerberosConfigFileLocation - The location of the kerberos config file (on the server)
- * @param replicationFactor - dfs replication factor
- */
- public KerberosConfiguration(List<String> namedNodes,
- String kerberosConfigFileLocation,
- int replicationFactor) {
-
- setKerberosConfigFile(kerberosConfigFileLocation);
- setupHdfsConfiguration(namedNodes, replicationFactor);
- }
-
- private void setupHdfsConfiguration(List<String> namedNodes, int replicationFactor) {
- this.set(AUTHENTICATION_MODE, "kerberos");
-
- this.set(DFSConfigKeys.DFS_REPLICATION_KEY, Integer.toString(replicationFactor));
- this.set(DFSConfigKeys.DFS_NAMESERVICES, HFDS_NAMED_SERVICE);
- this.set(
- DFSUtil.addKeySuffixes(DFS_HA_NAMENODES_KEY_PREFIX, HFDS_NAMED_SERVICE),
- nodeToString(namedNodes.stream().map(this::nodeToString).collect(Collectors.joining(",")))
- );
-
- namedNodes.forEach(nodeName ->
- this.set(
- DFSUtil.addKeySuffixes(DFS_NAMENODE_RPC_ADDRESS_KEY, HFDS_NAMED_SERVICE, nodeToString(nodeName)),
- nodeName)
- );
-
- this.set(DFS_CLIENT_FAILOVER_PROXY_PROVIDER_KEY_PREFIX + "." + HFDS_NAMED_SERVICE, ConfiguredFailoverProxyProvider.class.getName());
-
- this.set(HFDS_FS, "hdfs://" + HFDS_NAMED_SERVICE);
- }
-
- /**
- * In order to connect to a hadoop cluster using Kerberos you need to add your own filesystem to the cache of the FileSystem component.
- * This is done by setting the uri that you use in your camel route as the URI that is used to setup the connection.
- * The URI is used as key when adding it to the cache (default functionality of the static FileSystem.get(URI, Configuration) method).
- *
- * @param username - Principal used to connect to the cluster
- * @param keyTabFileLocation - KeyTab file location (must be on the server)
- * @throws IOException - In case of error
- */
- public void loginWithKeytab(String username, String keyTabFileLocation) throws IOException {
- if (!new File(keyTabFileLocation).exists()) {
- throw new FileNotFoundException(format("KeyTab file [%s] could not be found.", keyTabFileLocation));
- }
- // we need to log in otherwise you cannot connect to the filesystem later on
- UserGroupInformation.setConfiguration(this);
- UserGroupInformation.loginUserFromKeytab(username, keyTabFileLocation);
- }
-
- /**
- * To use kerberos authentication, set the value of the 'java.security.krb5.conf' environment variable to an existing file.
- * If the environment variable is already set, warn if different than the specified parameter
- *
- * @param kerberosConfigFileLocation - kerb5.conf file (https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html)
- */
- public static void setKerberosConfigFile(String kerberosConfigFileLocation) {
- if (!new File(kerberosConfigFileLocation).exists()) {
- LOG.warn("Kerberos configuration file [{}}] could not be found.", kerberosConfigFileLocation);
- return;
- }
-
- String krb5Conf = System.getProperty(KERBEROS_5_SYS_ENV);
- if (krb5Conf == null || !krb5Conf.isEmpty()) {
- System.setProperty(KERBEROS_5_SYS_ENV, kerberosConfigFileLocation);
- } else if (!krb5Conf.equalsIgnoreCase(kerberosConfigFileLocation)) {
- LOG.warn("[{}] was already configured with: [{}] config file", KERBEROS_5_SYS_ENV, krb5Conf);
- }
- }
-
- private String nodeToString(String nodeName) {
- return nodeName.replaceAll(":[0-9]*", "").replaceAll("\\.", "_");
- }
-
-}
diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationBuilder.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationBuilder.java
new file mode 100644
index 0000000..b5025cc
--- /dev/null
+++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationBuilder.java
@@ -0,0 +1,67 @@
+/*
+ * 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.camel.component.hdfs.kerberos;
+
+import java.io.File;
+
+import org.apache.hadoop.conf.Configuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class KerberosConfigurationBuilder {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KerberosConfigurationBuilder.class);
+
+ private static final String KERBEROS_5_SYS_ENV = "java.security.krb5.conf";
+ private static final String AUTHENTICATION_MODE = "hadoop.security.authentication";
+
+ private KerberosConfigurationBuilder() {
+ // hidden
+ }
+
+ /**
+ * Add all the kerberos specific settings needed for this authentication mode
+ *
+ * @param kerberosConfigFileLocation - The location of the kerberos config file (on the server)
+ */
+ public static void withKerberosConfiguration(Configuration configuration, String kerberosConfigFileLocation) {
+ setKerberosConfigFile(kerberosConfigFileLocation);
+ configuration.set(AUTHENTICATION_MODE, "kerberos");
+
+ }
+
+ /**
+ * To use kerberos authentication, set the value of the 'java.security.krb5.conf' environment variable to an existing file.
+ * If the environment variable is already set, warn if different than the specified parameter
+ *
+ * @param kerberosConfigFileLocation - kerb5.conf file (https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html)
+ */
+ public static void setKerberosConfigFile(String kerberosConfigFileLocation) {
+ if (!new File(kerberosConfigFileLocation).exists()) {
+ LOG.warn("Kerberos configuration file [{}}] could not be found.", kerberosConfigFileLocation);
+ return;
+ }
+
+ String krb5Conf = System.getProperty(KERBEROS_5_SYS_ENV);
+ if (krb5Conf == null || !krb5Conf.isEmpty()) {
+ System.setProperty(KERBEROS_5_SYS_ENV, kerberosConfigFileLocation);
+ } else if (!krb5Conf.equalsIgnoreCase(kerberosConfigFileLocation)) {
+ LOG.warn("[{}] was already configured with: [{}] config file", KERBEROS_5_SYS_ENV, krb5Conf);
+ }
+ }
+
+}
diff --git a/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/HaConfigurationBuilderTest.java b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/HaConfigurationBuilderTest.java
new file mode 100644
index 0000000..3cf20fc
--- /dev/null
+++ b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/HaConfigurationBuilderTest.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.hdfs;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.junit.Assert.assertThat;
+
+public class HaConfigurationBuilderTest {
+
+ @Test
+ public void withClusterConfiguration() {
+ // given
+ Configuration configuration = new Configuration();
+ List<String> namedNodes = Arrays.asList("kerb_node_01.example.com:8021", "kerb_node_02.example.com:8022");
+ int replicationFactor = 3;
+
+ // when
+ HaConfigurationBuilder.withClusterConfiguration(configuration, namedNodes, replicationFactor);
+
+ // then
+ assertThat(configuration, notNullValue());
+ assertThat(configuration.get(DFSConfigKeys.DFS_REPLICATION_KEY), is("3"));
+ assertThat(configuration.get(DFSConfigKeys.DFS_NAMESERVICES), is("hfdsNamedService"));
+ assertThat(configuration.get("dfs.ha.namenodes.hfdsNamedService"), is("kerb_node_01_example_com,kerb_node_02_example_com"));
+ assertThat(configuration.get("dfs.namenode.rpc-address.hfdsNamedService.kerb_node_01_example_com"), is("kerb_node_01.example.com:8021"));
+ assertThat(configuration.get("dfs.namenode.rpc-address.hfdsNamedService.kerb_node_02_example_com"), is("kerb_node_02.example.com:8022"));
+ assertThat(configuration.get("dfs.client.failover.proxy.provider.hfdsNamedService"), is("org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"));
+ assertThat(configuration.get("fs.defaultFS"), is("hdfs://hfdsNamedService"));
+ }
+
+}
diff --git a/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosAuthenticationTest.java b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosAuthenticationTest.java
new file mode 100644
index 0000000..ca94682
--- /dev/null
+++ b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosAuthenticationTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.camel.component.hdfs.kerberos;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+
+public class KerberosAuthenticationTest {
+
+ private KerberosAuthentication underTest;
+
+ @Test
+ public void loginWithKeytabFile() throws IOException {
+ // given
+ Configuration configuration = new Configuration();
+
+ String username = "test_user";
+ String keyTabFileLocation = pwd() + "/src/test/resources/kerberos/test-keytab.bin";
+
+ underTest = new KerberosAuthentication(configuration, username, keyTabFileLocation);
+
+ // when
+ underTest.loginWithKeytab();
+
+ // then
+ /* message is printed in the logs */
+ }
+
+ @Test(expected = FileNotFoundException.class)
+ public void loginWithMissingKeytabFile() throws IOException {
+ // given
+ Configuration configuration = new Configuration();
+
+ String username = "test_user";
+ String keyTabFileLocation = pwd() + "/src/test/resources/kerberos/missing.bin";
+
+ underTest = new KerberosAuthentication(configuration, username, keyTabFileLocation);
+
+ // when
+ underTest.loginWithKeytab();
+
+ // then
+ /* exception was thrown */
+ }
+
+ private String pwd() {
+ return new File(".").getAbsolutePath();
+ }
+
+}
diff --git a/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationBuilderTest.java b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationBuilderTest.java
new file mode 100644
index 0000000..1cad1da
--- /dev/null
+++ b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationBuilderTest.java
@@ -0,0 +1,59 @@
+package org.apache.camel.component.hdfs.kerberos;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class KerberosConfigurationBuilderTest {
+
+ @Test
+ public void withKerberosConfiguration() {
+ // given
+ String kerberosConfigFileLocation = pwd() + "/src/test/resources/kerberos/test-kerb5.conf";
+
+ // when
+ KerberosConfigurationBuilder.setKerberosConfigFile(kerberosConfigFileLocation);
+
+ // then
+
+ }
+
+ @Test
+ public void setKerberosConfigFileWithRealFile() {
+ // given
+ String kerb5FileName = "test-kerb5.conf";
+ String kerberosConfigFileLocation = pwd() + "/src/test/resources/kerberos/" + kerb5FileName;
+
+ // when
+ KerberosConfigurationBuilder.setKerberosConfigFile(kerberosConfigFileLocation);
+
+ // then
+ String actual = System.getProperty("java.security.krb5.conf");
+ assertNotNull(actual);
+ assertTrue(actual.endsWith(kerb5FileName));
+ }
+
+ @Test
+ public void setKerberosConfigFileWithMissingFile() {
+ // given
+ String kerb5FileName = "missing-kerb5.conf";
+ String kerberosConfigFileLocation = pwd() + "/src/test/resources/kerberos/" + kerb5FileName;
+
+ // when
+ KerberosConfigurationBuilder.setKerberosConfigFile(kerberosConfigFileLocation);
+
+ // then
+ String actual = System.getProperty("java.security.krb5.conf");
+ assertNull(actual);
+ }
+
+ private String pwd() {
+ return new File(".").getAbsolutePath();
+ }
+
+}
diff --git a/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationTest.java b/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationTest.java
deleted file mode 100644
index 9c9678b..0000000
--- a/components/camel-hdfs/src/test/java/org/apache/camel/component/hdfs/kerberos/KerberosConfigurationTest.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.camel.component.hdfs.kerberos;
-
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
-
-import org.apache.hadoop.hdfs.DFSConfigKeys;
-import org.junit.Test;
-
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.CoreMatchers.notNullValue;
-import static org.junit.Assert.assertThat;
-
-public class KerberosConfigurationTest {
-
- private KerberosConfiguration underTest;
-
- @Test
- public void newKerberosConfiguration() {
- // given
- List<String> namedNodes = Arrays.asList("kerb_node_01.example.com:8021", "kerb_node_02.example.com:8022");
- String kerberosConfigFileLocation = pwd() + "/src/test/resources/kerberos/test-kerb5.conf";
- int replicationFactor = 3;
-
- // when
- underTest = new KerberosConfiguration(namedNodes, kerberosConfigFileLocation, replicationFactor);
-
- // then
- assertThat(underTest, notNullValue());
- assertThat(underTest.get("hadoop.security.authentication"), is("kerberos"));
- assertThat(underTest.get(DFSConfigKeys.DFS_REPLICATION_KEY), is("3"));
- assertThat(underTest.get(DFSConfigKeys.DFS_NAMESERVICES), is("hfdsNamedService"));
- assertThat(underTest.get("dfs.ha.namenodes.hfdsNamedService"), is("kerb_node_01_example_com,kerb_node_02_example_com"));
- assertThat(underTest.get("dfs.namenode.rpc-address.hfdsNamedService.kerb_node_01_example_com"), is("kerb_node_01.example.com:8021"));
- assertThat(underTest.get("dfs.namenode.rpc-address.hfdsNamedService.kerb_node_02_example_com"), is("kerb_node_02.example.com:8022"));
- assertThat(underTest.get("dfs.client.failover.proxy.provider.hfdsNamedService"), is("org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"));
- assertThat(underTest.get("fs.defaultFS"), is("hdfs://hfdsNamedService"));
- }
-
- @Test
- public void newKerberosConfigurationWithMissingKerberosConfigFile() {
- // given
- List<String> namedNodes = Arrays.asList("kerb_node_01.example.com:8021", "kerb_node_02.example.com:8022");
- String kerberosConfigFileLocation = pwd() + "/src/test/resources/kerberos/missing.conf";
- int replicationFactor = 3;
-
- // when
- underTest = new KerberosConfiguration(namedNodes, kerberosConfigFileLocation, replicationFactor);
-
- // then
- /* message is printed in the logs */
- }
-
- @Test(expected = FileNotFoundException.class)
- public void loginWithMissingKeytabFile() throws IOException {
- // given
- List<String> namedNodes = Arrays.asList("kerb_node_01.example.com:8021", "kerb_node_02.example.com:8022");
- String kerberosConfigFileLocation = pwd() + "/src/test/resources/kerberos/test-kerb5.conf";
- int replicationFactor = 3;
- underTest = new KerberosConfiguration(namedNodes, kerberosConfigFileLocation, replicationFactor);
-
- String username = "test_user";
- String keyTabFileLocation = pwd() + "/src/test/resources/kerberos/missing.bin";
-
- // when
- underTest.loginWithKeytab(username, keyTabFileLocation);
-
- // then
- /* exception was thrown */
- }
-
- private String pwd() {
- return new File(".").getAbsolutePath();
- }
-
-}
\ No newline at end of file
diff --git a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/HdfsEndpointBuilderFactory.java b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/HdfsEndpointBuilderFactory.java
index d824b02..c3aeb51 100644
--- a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/HdfsEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/HdfsEndpointBuilderFactory.java
@@ -159,20 +159,6 @@ public interface HdfsEndpointBuilderFactory {
return this;
}
/**
- * A comma separated list of kerberos nodes (e.g.
- * srv11.example.com:8021,srv12.example.com:8021) - see kerb5.conf file
- * (https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html).
- *
- * The option is a: <code>java.lang.String</code> type.
- *
- * Group: common
- */
- default HdfsEndpointConsumerBuilder kerberosNamedNodes(
- String kerberosNamedNodes) {
- doSetProperty("kerberosNamedNodes", kerberosNamedNodes);
- return this;
- }
- /**
* The username used to authenticate with the kerberos nodes.
*
* The option is a: <code>java.lang.String</code> type.
@@ -209,6 +195,18 @@ public interface HdfsEndpointBuilderFactory {
return this;
}
/**
+ * A comma separated list of named nodes (e.g.
+ * srv11.example.com:8020,srv12.example.com:8020).
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default HdfsEndpointConsumerBuilder namedNodes(String namedNodes) {
+ doSetProperty("namedNodes", namedNodes);
+ return this;
+ }
+ /**
* The file owner must match this owner for the consumer to pickup the
* file. Otherwise the file is skipped.
*
@@ -1171,20 +1169,6 @@ public interface HdfsEndpointBuilderFactory {
return this;
}
/**
- * A comma separated list of kerberos nodes (e.g.
- * srv11.example.com:8021,srv12.example.com:8021) - see kerb5.conf file
- * (https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html).
- *
- * The option is a: <code>java.lang.String</code> type.
- *
- * Group: common
- */
- default HdfsEndpointProducerBuilder kerberosNamedNodes(
- String kerberosNamedNodes) {
- doSetProperty("kerberosNamedNodes", kerberosNamedNodes);
- return this;
- }
- /**
* The username used to authenticate with the kerberos nodes.
*
* The option is a: <code>java.lang.String</code> type.
@@ -1221,6 +1205,18 @@ public interface HdfsEndpointBuilderFactory {
return this;
}
/**
+ * A comma separated list of named nodes (e.g.
+ * srv11.example.com:8020,srv12.example.com:8020).
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default HdfsEndpointProducerBuilder namedNodes(String namedNodes) {
+ doSetProperty("namedNodes", namedNodes);
+ return this;
+ }
+ /**
* The file owner must match this owner for the consumer to pickup the
* file. Otherwise the file is skipped.
*
@@ -1749,19 +1745,6 @@ public interface HdfsEndpointBuilderFactory {
return this;
}
/**
- * A comma separated list of kerberos nodes (e.g.
- * srv11.example.com:8021,srv12.example.com:8021) - see kerb5.conf file
- * (https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html).
- *
- * The option is a: <code>java.lang.String</code> type.
- *
- * Group: common
- */
- default HdfsEndpointBuilder kerberosNamedNodes(String kerberosNamedNodes) {
- doSetProperty("kerberosNamedNodes", kerberosNamedNodes);
- return this;
- }
- /**
* The username used to authenticate with the kerberos nodes.
*
* The option is a: <code>java.lang.String</code> type.
@@ -1797,6 +1780,18 @@ public interface HdfsEndpointBuilderFactory {
return this;
}
/**
+ * A comma separated list of named nodes (e.g.
+ * srv11.example.com:8020,srv12.example.com:8020).
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default HdfsEndpointBuilder namedNodes(String namedNodes) {
+ doSetProperty("namedNodes", namedNodes);
+ return this;
+ }
+ /**
* The file owner must match this owner for the consumer to pickup the
* file. Otherwise the file is skipped.
*