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 2016/08/05 06:24:34 UTC
[2/2] camel git commit: camel-kafka cleanup options and add missing
option.
camel-kafka cleanup options and add missing option.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/46d3e80a
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/46d3e80a
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/46d3e80a
Branch: refs/heads/master
Commit: 46d3e80af69a6d82d8c8b22eb18373e3f99dcd45
Parents: b87baca
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Aug 5 08:24:13 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Aug 5 08:24:13 2016 +0200
----------------------------------------------------------------------
components/camel-kafka/src/main/docs/kafka.adoc | 47 +++---
.../component/kafka/KafkaConfiguration.java | 149 +++++++++++--------
2 files changed, 115 insertions(+), 81 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/46d3e80a/components/camel-kafka/src/main/docs/kafka.adoc
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/docs/kafka.adoc b/components/camel-kafka/src/main/docs/kafka.adoc
index d8753d2..2c015f9 100644
--- a/components/camel-kafka/src/main/docs/kafka.adoc
+++ b/components/camel-kafka/src/main/docs/kafka.adoc
@@ -93,8 +93,9 @@ The Kafka component supports 1 options which are listed below.
+
// endpoint options: START
-The Kafka component supports 76 endpoint options which are listed below:
+The Kafka component supports 77 endpoint options which are listed below:
{% raw %}
[width="100%",cols="2s,1,1m,1m,5",options="header"]
@@ -103,12 +104,7 @@ The Kafka component supports 76 endpoint options which are listed below:
| brokers | common | | String | *Required* This is for bootstrapping and the producer will only use it for getting metadata (topics partitions and replicas). The socket connections for sending the actual data will be established based on the broker information returned in the metadata. The format is host1:port1host2:port2 and the list can be a subset of brokers or a VIP pointing to a subset of brokers. This option is known as metadata.broker.list in the Kafka documentation.
| clientId | common | | String | The client id is a user-specified string sent in each request to help trace calls. It should logically identify the application making the request.
| groupId | common | | String | A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group id multiple processes indicate that they are all part of the same consumer group.
-| kerberosBeforeReloginMinTime | common | 60000 | Integer | Login thread sleep time between refresh attempts.
-| kerberosInitCmd | common | /usr/bin/kinit | String | Kerberos kinit command path. Default is /usr/bin/kinit
-| kerberosRenewJitter | common | 0.05 | Double | Percentage of random jitter added to the renewal time.
-| kerberosRenewWindowFactor | common | 0.8 | Double | Login thread will sleep until the specified window factor of time from last refresh to ticket's expiry has been reached at which time it will try to renew the ticket.
| partitioner | common | org.apache.kafka.clients.producer.internals.DefaultPartitioner | String | The partitioner class for partitioning messages amongst sub-topics. The default partitioner is based on the hash of the key.
-| saslMechanism | common | GSSAPI | String | The Simple Authentication and Security Layer (SASL) Mechanism used. For the valid values see http://www.iana.org/assignments/sasl-mechanisms/sasl-mechanisms.xhtml
| topic | common | | String | *Required* Name of the topic to use.
| autoCommitEnable | consumer | true | Boolean | If true periodically commit to ZooKeeper the offset of messages already fetched by the consumer. This committed offset will be used when the process fails as the position from which the new consumer will begin.
| autoCommitIntervalMs | consumer | 5000 | Integer | The frequency in ms that the consumer offsets are committed to zookeeper.
@@ -153,29 +149,35 @@ The Kafka component supports 76 endpoint options which are listed below:
| requestTimeoutMs | producer | 30000 | Integer | The amount of time the broker will wait trying to meet the request.required.acks requirement before sending back an error to the client.
| retries | producer | 0 | Integer | Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error. Note that this retry is no different than if the client resent the record upon receiving the error. Allowing retries will potentially change the ordering of records because if two records are sent to a single partition and the first fails and is retried but the second succeeds then the second record may appear first.
| retryBackoffMs | producer | 100 | Integer | Before each retry the producer refreshes the metadata of relevant topics to see if a new leader has been elected. Since leader election takes a bit of time this property specifies the amount of time that the producer waits before refreshing the metadata.
-| saslKerberosServiceName | producer | | String | The Kerberos principal name that Kafka runs as. This can be defined either in Kafka's JAAS config or in Kafka's config.
-| securityProtocol | producer | PLAINTEXT | String | Protocol used to communicate with brokers. Currently only PLAINTEXT and SSL are supported.
| sendBufferBytes | producer | 131072 | Integer | Socket write buffer size
| serializerClass | producer | | String | The serializer class for messages. The default encoder takes a byte and returns the same byte. The default class is kafka.serializer.DefaultEncoder
-| sslCipherSuites | producer | | String | A list of cipher suites. This is a named combination of authentication encryption MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol.By default all the available cipher suites are supported.
-| sslEnabledProtocols | producer | TLSv1.2,TLSv1.1,TLSv1 | String | The list of protocols enabled for SSL connections. TLSv1.2 TLSv1.1 and TLSv1 are enabled by default.
-| sslEndpointAlgorithm | producer | | String | The endpoint identification algorithm to validate server hostname using server certificate.
-| sslKeymanagerAlgorithm | producer | SunX509 | String | The algorithm used by key manager factory for SSL connections. Default value is the key manager factory algorithm configured for the Java Virtual Machine.
-| sslKeyPassword | producer | | String | The password of the private key in the key store file. This is optional for client.
-| sslKeystoreLocation | producer | | String | The location of the key store file. This is optional for client and can be used for two-way authentication for client.
-| sslKeystorePassword | producer | | String | The store password for the key store file.This is optional for client and only needed if ssl.keystore.location is configured.
-| sslKeystoreType | producer | JKS | String | The file format of the key store file. This is optional for client. Default value is JKS
-| sslProtocol | producer | TLS | String | The SSL protocol used to generate the SSLContext. Default setting is TLS which is fine for most cases. Allowed values in recent JVMs are TLS TLSv1.1 and TLSv1.2. SSL SSLv2 and SSLv3 may be supported in older JVMs but their usage is discouraged due to known security vulnerabilities.
-| sslProvider | producer | | String | The name of the security provider used for SSL connections. Default value is the default security provider of the JVM.
-| sslTrustmanagerAlgorithm | producer | PKIX | String | The algorithm used by trust manager factory for SSL connections. Default value is the trust manager factory algorithm configured for the Java Virtual Machine.
-| sslTruststoreLocation | producer | | String | The location of the trust store file.
-| sslTruststorePassword | producer | | String | The password for the trust store file.
-| sslTruststoreType | producer | JKS | String | The file format of the trust store file. Default value is JKS.
| workerPool | producer | | ExecutorService | To use a custom worker pool for continue routing Exchange after kafka server has acknowledge the message that was sent to it from KafkaProducer using asynchronous non-blocking processing.
| workerPoolCoreSize | producer | 10 | Integer | Number of core threads for the worker pool for continue routing Exchange after kafka server has acknowledge the message that was sent to it from KafkaProducer using asynchronous non-blocking processing.
| workerPoolMaxSize | producer | 20 | Integer | Maximum number of threads for the worker pool for continue routing Exchange after kafka server has acknowledge the message that was sent to it from KafkaProducer using asynchronous non-blocking processing.
| exchangePattern | advanced | InOnly | ExchangePattern | Sets the default exchange pattern when creating an exchange
| synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
+| kerberosBeforeReloginMinTime | security | 60000 | Integer | Login thread sleep time between refresh attempts.
+| kerberosInitCmd | security | /usr/bin/kinit | String | Kerberos kinit command path. Default is /usr/bin/kinit
+| kerberosPrincipalToLocalRules | security | DEFAULT | String | A list of rules for mapping from principal names to short names (typically operating system usernames). The rules are evaluated in order and the first rule that matches a principal name is used to map it to a short name. Any later rules in the list are ignored. By default principal names of the form username/hostnameREALM are mapped to username. For more details on the format please see security authorization and acls. Multiple values can be separated by comma
+| kerberosRenewJitter | security | 0.05 | Double | Percentage of random jitter added to the renewal time.
+| kerberosRenewWindowFactor | security | 0.8 | Double | Login thread will sleep until the specified window factor of time from last refresh to ticket's expiry has been reached at which time it will try to renew the ticket.
+| saslKerberosServiceName | security | | String | The Kerberos principal name that Kafka runs as. This can be defined either in Kafka's JAAS config or in Kafka's config.
+| saslMechanism | security | GSSAPI | String | The Simple Authentication and Security Layer (SASL) Mechanism used. For the valid values see http://www.iana.org/assignments/sasl-mechanisms/sasl-mechanisms.xhtml
+| securityProtocol | security | PLAINTEXT | String | Protocol used to communicate with brokers. Currently only PLAINTEXT and SSL are supported.
+| sslCipherSuites | security | | String | A list of cipher suites. This is a named combination of authentication encryption MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol.By default all the available cipher suites are supported.
+| sslEnabledProtocols | security | TLSv1.2,TLSv1.1,TLSv1 | String | The list of protocols enabled for SSL connections. TLSv1.2 TLSv1.1 and TLSv1 are enabled by default.
+| sslEndpointAlgorithm | security | | String | The endpoint identification algorithm to validate server hostname using server certificate.
+| sslKeymanagerAlgorithm | security | SunX509 | String | The algorithm used by key manager factory for SSL connections. Default value is the key manager factory algorithm configured for the Java Virtual Machine.
+| sslKeyPassword | security | | String | The password of the private key in the key store file. This is optional for client.
+| sslKeystoreLocation | security | | String | The location of the key store file. This is optional for client and can be used for two-way authentication for client.
+| sslKeystorePassword | security | | String | The store password for the key store file.This is optional for client and only needed if ssl.keystore.location is configured.
+| sslKeystoreType | security | JKS | String | The file format of the key store file. This is optional for client. Default value is JKS
+| sslProtocol | security | TLS | String | The SSL protocol used to generate the SSLContext. Default setting is TLS which is fine for most cases. Allowed values in recent JVMs are TLS TLSv1.1 and TLSv1.2. SSL SSLv2 and SSLv3 may be supported in older JVMs but their usage is discouraged due to known security vulnerabilities.
+| sslProvider | security | | String | The name of the security provider used for SSL connections. Default value is the default security provider of the JVM.
+| sslTrustmanagerAlgorithm | security | PKIX | String | The algorithm used by trust manager factory for SSL connections. Default value is the trust manager factory algorithm configured for the Java Virtual Machine.
+| sslTruststoreLocation | security | | String | The location of the trust store file.
+| sslTruststorePassword | security | | String | The password for the trust store file.
+| sslTruststoreType | security | JKS | String | The file format of the trust store file. Default value is JKS.
|=======================================================================
{% endraw %}
// endpoint options: END
@@ -195,6 +197,7 @@ The Kafka component supports 76 endpoint options which are listed below:
+
For more information about Producer/Consumer configuration:
http://kafka.apache.org/documentation.html#newconsumerconfigs[http://kafka.apache.org/documentation.html#newconsumerconfigs]
http://git-wip-us.apache.org/repos/asf/camel/blob/46d3e80a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
index deb12a1..db7f4dc 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.kafka;
+import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
@@ -132,22 +133,6 @@ public class KafkaConfiguration {
//retries
@UriParam(label = "producer", defaultValue = "0")
private Integer retries = 0;
- // SSL
- // ssl.key.password
- @UriParam(label = "producer")
- private String sslKeyPassword;
- // ssl.keystore.location
- @UriParam(label = "producer")
- private String sslKeystoreLocation;
- // ssl.keystore.password
- @UriParam(label = "producer")
- private String sslKeystorePassword;
- //ssl.truststore.location
- @UriParam(label = "producer")
- private String sslTruststoreLocation;
- //ssl.truststore.password
- @UriParam(label = "producer")
- private String sslTruststorePassword;
//batch.size
@UriParam(label = "producer", defaultValue = "16384")
private Integer producerBatchSize = 16384;
@@ -169,32 +154,11 @@ public class KafkaConfiguration {
//request.timeout.ms
@UriParam(label = "producer", defaultValue = "30000")
private Integer requestTimeoutMs = 30000;
- // SASL & sucurity Protocol
- //sasl.kerberos.service.name
- @UriParam(label = "producer")
- private String saslKerberosServiceName;
- //security.protocol
- @UriParam(label = "producer", defaultValue = CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL)
- private String securityProtocol = CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL;
//send.buffer.bytes
@UriParam(label = "producer", defaultValue = "131072")
private Integer sendBufferBytes = 131072;
- //SSL
- //ssl.enabled.protocols
- @UriParam(label = "producer", defaultValue = SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS)
- private String sslEnabledProtocols = SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS;
- //ssl.keystore.type
- @UriParam(label = "producer", defaultValue = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE)
- private String sslKeystoreType = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE;
- //ssl.protocol
- @UriParam(label = "producer", defaultValue = SslConfigs.DEFAULT_SSL_PROTOCOL)
- private String sslProtocol = SslConfigs.DEFAULT_SSL_PROTOCOL;
- //ssl.provider
- @UriParam(label = "producer")
- private String sslProvider;
- //ssl.truststore.type
- @UriParam(label = "producer", defaultValue = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE)
- private String sslTruststoreType = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE;
+ @UriParam(label = "producer", defaultValue = "true")
+ private boolean recordMetadata = true;
//max.in.flight.requests.per.connection
@UriParam(label = "producer", defaultValue = "5")
private Integer maxInFlightRequest = 5;
@@ -213,37 +177,77 @@ public class KafkaConfiguration {
//reconnect.backoff.ms
@UriParam(label = "producer", defaultValue = "50")
private Integer reconnectBackoffMs = 50;
+ // SSL
+ // ssl.key.password
+ @UriParam(label = "producer,security")
+ private String sslKeyPassword;
+ // ssl.keystore.location
+ @UriParam(label = "producer,security")
+ private String sslKeystoreLocation;
+ // ssl.keystore.password
+ @UriParam(label = "producer,security")
+ private String sslKeystorePassword;
+ //ssl.truststore.location
+ @UriParam(label = "producer,security")
+ private String sslTruststoreLocation;
+ //ssl.truststore.password
+ @UriParam(label = "producer,security")
+ private String sslTruststorePassword;
+ //SSL
+ //ssl.enabled.protocols
+ @UriParam(label = "common,security", defaultValue = SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS)
+ private String sslEnabledProtocols = SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS;
+ //ssl.keystore.type
+ @UriParam(label = "common,security", defaultValue = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE)
+ private String sslKeystoreType = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE;
+ //ssl.protocol
+ @UriParam(label = "common,security", defaultValue = SslConfigs.DEFAULT_SSL_PROTOCOL)
+ private String sslProtocol = SslConfigs.DEFAULT_SSL_PROTOCOL;
+ //ssl.provider
+ @UriParam(label = "common,security")
+ private String sslProvider;
+ //ssl.truststore.type
+ @UriParam(label = "common,security", defaultValue = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE)
+ private String sslTruststoreType = SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE;
+ //SSL
+ //ssl.cipher.suites
+ @UriParam(label = "common,security")
+ private String sslCipherSuites;
+ //ssl.endpoint.identification.algorithm
+ @UriParam(label = "common,security")
+ private String sslEndpointAlgorithm;
+ //ssl.keymanager.algorithm
+ @UriParam(label = "common,security", defaultValue = "SunX509")
+ private String sslKeymanagerAlgorithm = "SunX509";
+ //ssl.trustmanager.algorithm
+ @UriParam(label = "common,security", defaultValue = "PKIX")
+ private String sslTrustmanagerAlgorithm = "PKIX";
+ // SASL & sucurity Protocol
+ //sasl.kerberos.service.name
+ @UriParam(label = "common,security")
+ private String saslKerberosServiceName;
+ //security.protocol
+ @UriParam(label = "common,security", defaultValue = CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL)
+ private String securityProtocol = CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL;
//SASL
//sasl.kerberos.kinit.cmd
- @UriParam(label = "common", defaultValue = SaslConfigs.DEFAULT_SASL_MECHANISM)
+ @UriParam(label = "common,security", defaultValue = SaslConfigs.DEFAULT_SASL_MECHANISM)
private String saslMechanism = SaslConfigs.DEFAULT_SASL_MECHANISM;
//sasl.kerberos.kinit.cmd
- @UriParam(label = "common", defaultValue = SaslConfigs.DEFAULT_KERBEROS_KINIT_CMD)
+ @UriParam(label = "common,security", defaultValue = SaslConfigs.DEFAULT_KERBEROS_KINIT_CMD)
private String kerberosInitCmd = SaslConfigs.DEFAULT_KERBEROS_KINIT_CMD;
//sasl.kerberos.min.time.before.relogin
- @UriParam(label = "common", defaultValue = "60000")
+ @UriParam(label = "common,security", defaultValue = "60000")
private Integer kerberosBeforeReloginMinTime = 60000;
//sasl.kerberos.ticket.renew.jitter
- @UriParam(label = "common", defaultValue = "0.05")
+ @UriParam(label = "common,security", defaultValue = "0.05")
private Double kerberosRenewJitter = SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_JITTER;
//sasl.kerberos.ticket.renew.window.factor
- @UriParam(label = "common", defaultValue = "0.8")
+ @UriParam(label = "common,security", defaultValue = "0.8")
private Double kerberosRenewWindowFactor = SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_WINDOW_FACTOR;
- //SSL
- //ssl.cipher.suites
- @UriParam(label = "producer")
- private String sslCipherSuites;
- //ssl.endpoint.identification.algorithm
- @UriParam(label = "producer")
- private String sslEndpointAlgorithm;
- //ssl.keymanager.algorithm
- @UriParam(label = "producer", defaultValue = "SunX509")
- private String sslKeymanagerAlgorithm = "SunX509";
- //ssl.trustmanager.algorithm
- @UriParam(label = "producer", defaultValue = "PKIX")
- private String sslTrustmanagerAlgorithm = "PKIX";
- @UriParam(label = "producer", defaultValue = "true")
- private boolean recordMetadata = true;
+ @UriParam(label = "common,security", defaultValue = "DEFAULT")
+ //sasl.kerberos.principal.to.local.rules
+ private String kerberosPrincipalToLocalRules;
public KafkaConfiguration() {
}
@@ -294,6 +298,7 @@ public class KafkaConfiguration {
addPropertyIfNotNull(props, SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN, getKerberosBeforeReloginMinTime());
addPropertyIfNotNull(props, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER, getKerberosRenewJitter());
addPropertyIfNotNull(props, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, getKerberosRenewWindowFactor());
+ addListPropertyIfNotNull(props, SaslConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES, getKerberosPrincipalToLocalRules());
addPropertyIfNotNull(props, SaslConfigs.SASL_MECHANISM, getSaslMechanism());
//SSL
addPropertyIfNotNull(props, SslConfigs.SSL_CIPHER_SUITES_CONFIG, getSslCipherSuites());
@@ -350,6 +355,7 @@ public class KafkaConfiguration {
addPropertyIfNotNull(props, SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN, getKerberosBeforeReloginMinTime());
addPropertyIfNotNull(props, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER, getKerberosRenewJitter());
addPropertyIfNotNull(props, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, getKerberosRenewWindowFactor());
+ addListPropertyIfNotNull(props, SaslConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES, getKerberosPrincipalToLocalRules());
addPropertyIfNotNull(props, SaslConfigs.SASL_MECHANISM, getSaslMechanism());
//SSL
addPropertyIfNotNull(props, SslConfigs.SSL_CIPHER_SUITES_CONFIG, getSslCipherSuites());
@@ -366,6 +372,15 @@ public class KafkaConfiguration {
}
}
+ private static <T> void addListPropertyIfNotNull(Properties props, String key, T value) {
+ if (value != null) {
+ // Kafka expects all properties as String
+ String[] values = value.toString().split(",");
+ List<String> list = Arrays.asList(values);
+ props.put(key, list);
+ }
+ }
+
public String getGroupId() {
return groupId;
}
@@ -651,6 +666,22 @@ public class KafkaConfiguration {
this.kerberosRenewWindowFactor = kerberosRenewWindowFactor;
}
+ public String getKerberosPrincipalToLocalRules() {
+ return kerberosPrincipalToLocalRules;
+ }
+
+ /**
+ * A list of rules for mapping from principal names to short names (typically operating system usernames).
+ * The rules are evaluated in order and the first rule that matches a principal name is used to map it to a short name. Any later rules in the list are ignored.
+ * By default, principal names of the form {username}/{hostname}@{REALM} are mapped to {username}.
+ * For more details on the format please see <a href=\"#security_authz\"> security authorization and acls</a>.
+ * <p/>
+ * Multiple values can be separated by comma
+ */
+ public void setKerberosPrincipalToLocalRules(String kerberosPrincipalToLocalRules) {
+ this.kerberosPrincipalToLocalRules = kerberosPrincipalToLocalRules;
+ }
+
public String getSslCipherSuites() {
return sslCipherSuites;
}