From cc69f15acecbac4b3a0ff2f7884b90d39d97b3f5 Mon Sep 17 00:00:00 2001 From: Magnus Edenhill Date: Thu, 19 May 2016 20:19:20 +0200 Subject: [PATCH 1/4] sasl.kerberos.kinit.cmd now takes a command+args string template (issue #635) This is an incompatible change and will break existing users of Kerberos kinit who have chosen to modify the kinit command. Also, the keytab file is not automatically used when configured, the user needs to configure sasl.kerberos.kinit.cmd accordingly, e.g. by adding: ".. -k %{sasl.kerberos.keytab}" --- CONFIGURATION.md | 4 +- src/Makefile | 2 +- src/rd.h | 8 +++ src/rdkafka.c | 13 ++++ src/rdkafka_conf.c | 11 ++- src/rdkafka_sasl.c | 114 ++++++++++++++++++++++++------- src/rdkafka_sasl.h | 2 + src/rdstring.c | 140 +++++++++++++++++++++++++++++++++++++++ win32/librdkafka.vcxproj | 1 + 9 files changed, 265 insertions(+), 30 deletions(-) create mode 100644 src/rdstring.c diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 8103ff1ee3..fbe841a168 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -52,8 +52,8 @@ ssl.crl.location | * | | sasl.mechanisms | * | GSSAPI, PLAIN | GSSAPI | SASL mechanism to use for authentication. Supported: GSSAPI, PLAIN. **NOTE**: Despite the name only one mechanism must be configured.
*Type: string* sasl.kerberos.service.name | * | | kafka | Kerberos principal name that Kafka runs as.
*Type: string* sasl.kerberos.principal | * | | kafkaclient | This client's Kerberos principal name.
*Type: string* -sasl.kerberos.kinit.cmd | * | | kinit | Kerberos kinit command path.
*Type: string* -sasl.kerberos.keytab | * | | | Path to Kerberos keytab file. Uses system default if not set.
*Type: string* +sasl.kerberos.kinit.cmd | * | | kinit -S "%{sasl.kerberos.service.name}/%{broker.name}" -k -i %{sasl.kerberos.principal} | Full kerberos kinit command string, %{config.prop.name} is replaced by corresponding config object value, %{broker.name} returns the broker's hostname.
*Type: string* +sasl.kerberos.keytab | * | | | Path to Kerberos keytab file. Uses system default if not set.**NOTE**: This is not automatically used but must be added to the template in sasl.kerberos.kinit.cmd as ` ... -k %{sasl.kerberos.keytab}`.
*Type: string* sasl.kerberos.min.time.before.relogin | * | 1 .. 86400000 | 60000 | Minimum time in milliseconds between key refresh attempts.
*Type: integer* sasl.username | * | | | SASL username for use with the PLAIN mechanism
*Type: string* sasl.password | * | | | SASL password for use with the PLAIN mechanism
*Type: string* diff --git a/src/Makefile b/src/Makefile index e189638f23..217c5b0b4d 100644 --- a/src/Makefile +++ b/src/Makefile @@ -16,7 +16,7 @@ SRCS= rdkafka.c rdkafka_broker.c rdkafka_msg.c rdkafka_topic.c \ rdkafka_assignor.c rdkafka_range_assignor.c \ rdkafka_roundrobin_assignor.c rdkafka_feature.c \ rdcrc32.c rdaddr.c rdrand.c rdlist.c tinycthread.c \ - rdlog.c \ + rdlog.c rdstring.c \ $(SRCS_y) HDRS= rdkafka.h diff --git a/src/rd.h b/src/rd.h index d29096dd41..50adc01906 100644 --- a/src/rd.h +++ b/src/rd.h @@ -123,6 +123,14 @@ static RD_INLINE RD_UNUSED char *rd_strndup(const char *s, size_t len) { } +char *rd_string_render (const char *template, + char *errstr, size_t errstr_size, + ssize_t (*callback) (const char *key, + char *buf, size_t size, + void *opaque), + void *opaque); + + /* * Portability diff --git a/src/rdkafka.c b/src/rdkafka.c index 52d2a858a5..e22da970ba 100644 --- a/src/rdkafka.c +++ b/src/rdkafka.c @@ -1090,6 +1090,19 @@ rd_kafka_t *rd_kafka_new (rd_kafka_type_t type, rd_kafka_conf_t *conf, return NULL; } +#if WITH_SASL + if (rk->rk_conf.security_protocol == RD_KAFKA_PROTO_SASL_SSL || + rk->rk_conf.security_protocol == RD_KAFKA_PROTO_SASL_PLAINTEXT) { + /* Validate SASL config */ + if (rd_kafka_sasl_conf_validate(rk, errstr, errstr_size) == -1) { + rd_kafka_destroy_internal(rk); + rd_kafka_set_last_error(RD_KAFKA_RESP_ERR__INVALID_ARG, + EINVAL); + return NULL; + } + } +#endif + #if WITH_SSL if (rk->rk_conf.security_protocol == RD_KAFKA_PROTO_SSL || rk->rk_conf.security_protocol == RD_KAFKA_PROTO_SASL_SSL) { diff --git a/src/rdkafka_conf.c b/src/rdkafka_conf.c index 1f12973cac..b579096458 100644 --- a/src/rdkafka_conf.c +++ b/src/rdkafka_conf.c @@ -416,11 +416,16 @@ static const struct rd_kafka_property rd_kafka_properties[] = { .sdef = "kafkaclient" }, { _RK_GLOBAL, "sasl.kerberos.kinit.cmd", _RK_C_STR, _RK(sasl.kinit_cmd), - "Kerberos kinit command path.", - .sdef = "kinit" }, + "Full kerberos kinit command string, %{config.prop.name} is replaced " + "by corresponding config object value, %{broker.name} returns the " + "broker's hostname.", + .sdef = "kinit -S \"%{sasl.kerberos.service.name}/%{broker.name}\" -k -i %{sasl.kerberos.principal}" }, { _RK_GLOBAL, "sasl.kerberos.keytab", _RK_C_STR, _RK(sasl.keytab), - "Path to Kerberos keytab file. Uses system default if not set." }, + "Path to Kerberos keytab file. Uses system default if not set." + "**NOTE**: This is not automatically used but must be added to the " + "template in sasl.kerberos.kinit.cmd as " + "` ... -k %{sasl.kerberos.keytab}`." }, { _RK_GLOBAL, "sasl.kerberos.min.time.before.relogin", _RK_C_INT, _RK(sasl.relogin_min_time), "Minimum time in milliseconds between key refresh attempts.", diff --git a/src/rdkafka_sasl.c b/src/rdkafka_sasl.c index f68ec01466..4405fa0eed 100644 --- a/src/rdkafka_sasl.c +++ b/src/rdkafka_sasl.c @@ -210,6 +210,42 @@ int rd_kafka_sasl_io_event (rd_kafka_transport_t *rktrans, int events, +static ssize_t render_callback (const char *key, char *buf, + size_t size, void *opaque) { + rd_kafka_broker_t *rkb = opaque; + + if (!strcmp(key, "broker.name")) { + char *val, *t; + size_t len; + rd_kafka_broker_lock(rkb); + rd_strdupa(&val, rkb->rkb_nodename); + rd_kafka_broker_unlock(rkb); + + /* Just the broker name, no port */ + if ((t = strchr(val, ':'))) + len = (size_t)(t-val); + else + len = strlen(val); + + if (buf) + memcpy(buf, val, RD_MIN(len, size)); + + return len; + + } else { + rd_kafka_conf_res_t res; + size_t destsize = size; + + /* Try config lookup. */ + res = rd_kafka_conf_get(&rkb->rkb_rk->rk_conf, key, + buf, &destsize); + if (res != RD_KAFKA_CONF_OK) + return -1; + + /* Dont include \0 in returned size */ + return (destsize > 0 ? destsize-1 : destsize); + } +} /** @@ -222,33 +258,24 @@ int rd_kafka_sasl_io_event (rd_kafka_transport_t *rktrans, int events, static int rd_kafka_sasl_kinit_refresh (rd_kafka_broker_t *rkb) { rd_kafka_t *rk = rkb->rkb_rk; int r; - char cmd[512]; - char keytab[512]; - char *hostname, *t; + char *cmd; + char errstr[128]; - if (!rk->rk_conf.sasl.kinit_cmd || !strstr(rk->rk_conf.sasl.mechanisms, "GSSAPI")) + if (!rk->rk_conf.sasl.kinit_cmd || + !strstr(rk->rk_conf.sasl.mechanisms, "GSSAPI")) return 0; /* kinit not configured */ - /* Build kinit refresh command line for this broker. */ - rd_kafka_broker_lock(rkb); - rd_strdupa(&hostname, rkb->rkb_nodename); - rd_kafka_broker_unlock(rkb); - - if ((t = strchr(hostname, ':'))) - *t = '\0'; /* remove ":port" */ - - if (rk->rk_conf.sasl.keytab) - rd_snprintf(keytab, sizeof(keytab), - "-k -t \"%s\"", - rk->rk_conf.sasl.keytab); - else /* default path */ - rd_snprintf(keytab, sizeof(keytab), "-k -i"); - - rd_snprintf(cmd, sizeof(cmd), "%s -S \"%s/%s\" %s %s", - rk->rk_conf.sasl.kinit_cmd, - rk->rk_conf.sasl.service_name, hostname, - keytab, - rk->rk_conf.sasl.principal); + /* Build kinit refresh command line using string rendering and config */ + cmd = rd_string_render(rk->rk_conf.sasl.kinit_cmd, + errstr, sizeof(errstr), + render_callback, rkb); + if (!cmd) { + rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", + "Failed to construct kinit command " + "from sasl.kerberos.kinit.cmd template: %s", + errstr); + return -1; + } /* Execute kinit */ rd_rkb_dbg(rkb, SECURITY, "SASLREFRESH", @@ -259,19 +286,24 @@ static int rd_kafka_sasl_kinit_refresh (rd_kafka_broker_t *rkb) { rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", "SASL key refresh failed: Failed to execute %s", cmd); + rd_free(cmd); return -1; } else if (WIFSIGNALED(r)) { rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", "SASL key refresh failed: %s: received signal %d", cmd, WTERMSIG(r)); + rd_free(cmd); return -1; } else if (WIFEXITED(r) && WEXITSTATUS(r) != 0) { rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", "SASL key refresh failed: %s: exited with code %d", cmd, WEXITSTATUS(r)); + rd_free(cmd); return -1; } + rd_free(cmd); + rd_rkb_dbg(rkb, SECURITY, "SASLREFRESH", "SASL key refreshed"); return 0; } @@ -579,6 +611,40 @@ void rd_kafka_broker_sasl_init (rd_kafka_broker_t *rkb) { rd_kafka_sasl_kinit_refresh_tmr_cb, rkb); } + + +int rd_kafka_sasl_conf_validate (rd_kafka_t *rk, + char *errstr, size_t errstr_size) { + rd_kafka_broker_t rkb; + char *cmd; + char tmperr[128]; + + if (strcmp(rk->rk_conf.sasl.mechanisms, "GSSAPI")) + return 0; + + memset(&rkb, 0, sizeof(rkb)); + strcpy(rkb.rkb_nodename, "ATestBroker:9092"); + rkb.rkb_rk = rk; + mtx_init(&rkb.rkb_lock, mtx_plain); + + cmd = rd_string_render(rk->rk_conf.sasl.kinit_cmd, + tmperr, sizeof(tmperr), + render_callback, &rkb); + + mtx_destroy(&rkb.rkb_lock); + + if (!cmd) { + rd_snprintf(errstr, errstr_size, + "Invalid sasl.kerberos.kinit.cmd value: %s", + tmperr); + return -1; + } + + rd_free(cmd); + return 0; +} + + /** * Global SASL termination. * NOTE: Should not be called since the application may be using SASL too. diff --git a/src/rdkafka_sasl.h b/src/rdkafka_sasl.h index af59404eda..098cf54184 100644 --- a/src/rdkafka_sasl.h +++ b/src/rdkafka_sasl.h @@ -40,3 +40,5 @@ void rd_kafka_broker_sasl_init (rd_kafka_broker_t *rkb); int rd_kafka_sasl_global_init (void); +int rd_kafka_sasl_conf_validate (rd_kafka_t *rk, + char *errstr, size_t errstr_size); diff --git a/src/rdstring.c b/src/rdstring.c new file mode 100644 index 0000000000..91d3882817 --- /dev/null +++ b/src/rdstring.c @@ -0,0 +1,140 @@ +/* + * librdkafka - The Apache Kafka C/C++ library + * + * Copyright (c) 2016 Magnus Edenhill + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright notice, + * this list of conditions and the following disclaimer in the documentation + * and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + + +#include "rd.h" + +/** + * @brief Render string \p template using \p callback for key lookups. + * + * Keys in template follow the %{keyname} syntax. + * + * The \p callback must not write more than \p size bytes to \p buf, must + * should return the number of bytes it wanted to write (which will indicate + * a truncated write). + * If the key is not found -1 should be returned (which fails the rendering). + * + * @returns number of written bytes to \p dest, + * or -1 on failure (errstr is written) + */ +char *rd_string_render (const char *template, + char *errstr, size_t errstr_size, + ssize_t (*callback) (const char *key, + char *buf, size_t size, + void *opaque), + void *opaque) { + const char *s = template; + const char *tend = template + strlen(template); + size_t size = 256; + char *buf; + size_t of = 0; + + buf = rd_malloc(size); + +#define _remain() (size - of - 1) +#define _assure_space(SZ) do { \ + if (of + (SZ) + 1 >= size) { \ + size = (size + (SZ) + 1) * 2; \ + buf = realloc(buf, size); \ + } \ + } while (0) + +#define _do_write(PTR,SZ) do { \ + _assure_space(SZ); \ + memcpy(buf+of, (PTR), (SZ)); \ + of += (SZ); \ + } while (0) + + + + while (*s) { + const char *t; + size_t tof = (size_t)(s-template); + + t = strstr(s, "%{"); + if (t != s) { + /* Write "abc%{" + * ^^^ */ + size_t len = (size_t)((t ? t : tend)-s); + if (len) + _do_write(s, len); + } + + if (t) { + const char *te; + ssize_t r; + char *tmpkey; + + /* Find "abc%{key}" + * ^ */ + te = strchr(t+2, '}'); + if (!te) { + rd_snprintf(errstr, errstr_size, + "Missing close-brace } for " + "%.*s at %"PRIdsz, + 15, t, tof); + rd_free(buf); + return NULL; + } + + rd_strndupa(&tmpkey, t+2, (int)(te-t-2)); + + /* Query callback for length of key's value. */ + r = callback(tmpkey, NULL, 0, opaque); + if (r == -1) { + rd_snprintf(errstr, errstr_size, + "Property not available: \"%s\"", + tmpkey); + rd_free(buf); + return NULL; + } + + _assure_space(r); + + /* Call again now providing a large enough buffer. */ + r = callback(tmpkey, buf+of, _remain(), opaque); + if (r == -1) { + rd_snprintf(errstr, errstr_size, + "Property not available: " + "\"%s\"", tmpkey); + rd_free(buf); + return NULL; + } + + assert(r < (ssize_t)_remain()); + of += r; + s = te+1; + + } else { + s = tend; + } + } + + buf[of] = '\0'; + return buf; +} diff --git a/win32/librdkafka.vcxproj b/win32/librdkafka.vcxproj index 52a63ea8f9..69d1a6c02d 100644 --- a/win32/librdkafka.vcxproj +++ b/win32/librdkafka.vcxproj @@ -226,6 +226,7 @@ + From 51d7a3e69429f9f97610f35bc08bfecf48bdba42 Mon Sep 17 00:00:00 2001 From: Magnus Edenhill Date: Thu, 19 May 2016 20:20:18 +0200 Subject: [PATCH 2/4] Protect kinit execution by singleton lock to avoid cache race conditions (issue #635) --- src/rdkafka_sasl.c | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/rdkafka_sasl.c b/src/rdkafka_sasl.c index 4405fa0eed..f89066ec15 100644 --- a/src/rdkafka_sasl.c +++ b/src/rdkafka_sasl.c @@ -34,6 +34,8 @@ #include +static mtx_t rd_kafka_sasl_kinit_lock; + /** * Send auth message with framing. @@ -280,7 +282,10 @@ static int rd_kafka_sasl_kinit_refresh (rd_kafka_broker_t *rkb) { /* Execute kinit */ rd_rkb_dbg(rkb, SECURITY, "SASLREFRESH", "Refreshing SASL keys with command: %s", cmd); + + mtx_lock(&rd_kafka_sasl_kinit_lock); r = system(cmd); + mtx_unlock(&rd_kafka_sasl_kinit_lock); if (r == -1) { rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", @@ -651,17 +656,18 @@ int rd_kafka_sasl_conf_validate (rd_kafka_t *rk, */ void rd_kafka_sasl_global_term (void) { sasl_done(); + mtx_destroy(&rd_kafka_sasl_kinit_lock); } - - /** * Global SASL init, called once per runtime. */ int rd_kafka_sasl_global_init (void) { int r; + mtx_init(&rd_kafka_sasl_kinit_lock, mtx_plain); + r = sasl_client_init(NULL); if (r != SASL_OK) { fprintf(stderr, "librdkafka: sasl_client_init() failed: %s\n", From 62c551447c69d8d36529c55b9d9ca5b0c7955de9 Mon Sep 17 00:00:00 2001 From: Magnus Edenhill Date: Fri, 20 May 2016 11:41:52 +0200 Subject: [PATCH 3/4] Revert kinit.cmd logic to be backwards compatible --- CONFIGURATION.md | 4 ++-- src/rdkafka_conf.c | 5 +++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/CONFIGURATION.md b/CONFIGURATION.md index fbe841a168..77d9cde380 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -52,8 +52,8 @@ ssl.crl.location | * | | sasl.mechanisms | * | GSSAPI, PLAIN | GSSAPI | SASL mechanism to use for authentication. Supported: GSSAPI, PLAIN. **NOTE**: Despite the name only one mechanism must be configured.
*Type: string* sasl.kerberos.service.name | * | | kafka | Kerberos principal name that Kafka runs as.
*Type: string* sasl.kerberos.principal | * | | kafkaclient | This client's Kerberos principal name.
*Type: string* -sasl.kerberos.kinit.cmd | * | | kinit -S "%{sasl.kerberos.service.name}/%{broker.name}" -k -i %{sasl.kerberos.principal} | Full kerberos kinit command string, %{config.prop.name} is replaced by corresponding config object value, %{broker.name} returns the broker's hostname.
*Type: string* -sasl.kerberos.keytab | * | | | Path to Kerberos keytab file. Uses system default if not set.**NOTE**: This is not automatically used but must be added to the template in sasl.kerberos.kinit.cmd as ` ... -k %{sasl.kerberos.keytab}`.
*Type: string* +sasl.kerberos.kinit.cmd | * | | kinit -S "%{sasl.kerberos.service.name}/%{broker.name}" -k -t "%{sasl.kerberos.keytab}" %{sasl.kerberos.principal} | Full kerberos kinit command string, %{config.prop.name} is replaced by corresponding config object value, %{broker.name} returns the broker's hostname.
*Type: string* +sasl.kerberos.keytab | * | | | Path to Kerberos keytab file. Uses system default if not set.**NOTE**: This is not automatically used but must be added to the template in sasl.kerberos.kinit.cmd as ` ... -t %{sasl.kerberos.keytab}`.
*Type: string* sasl.kerberos.min.time.before.relogin | * | 1 .. 86400000 | 60000 | Minimum time in milliseconds between key refresh attempts.
*Type: integer* sasl.username | * | | | SASL username for use with the PLAIN mechanism
*Type: string* sasl.password | * | | | SASL password for use with the PLAIN mechanism
*Type: string* diff --git a/src/rdkafka_conf.c b/src/rdkafka_conf.c index b579096458..d62f640fe8 100644 --- a/src/rdkafka_conf.c +++ b/src/rdkafka_conf.c @@ -419,13 +419,14 @@ static const struct rd_kafka_property rd_kafka_properties[] = { "Full kerberos kinit command string, %{config.prop.name} is replaced " "by corresponding config object value, %{broker.name} returns the " "broker's hostname.", - .sdef = "kinit -S \"%{sasl.kerberos.service.name}/%{broker.name}\" -k -i %{sasl.kerberos.principal}" }, + .sdef = "kinit -S \"%{sasl.kerberos.service.name}/%{broker.name}\" " + "-k -t \"%{sasl.kerberos.keytab}\" %{sasl.kerberos.principal}" }, { _RK_GLOBAL, "sasl.kerberos.keytab", _RK_C_STR, _RK(sasl.keytab), "Path to Kerberos keytab file. Uses system default if not set." "**NOTE**: This is not automatically used but must be added to the " "template in sasl.kerberos.kinit.cmd as " - "` ... -k %{sasl.kerberos.keytab}`." }, + "` ... -t %{sasl.kerberos.keytab}`." }, { _RK_GLOBAL, "sasl.kerberos.min.time.before.relogin", _RK_C_INT, _RK(sasl.relogin_min_time), "Minimum time in milliseconds between key refresh attempts.", From 2213fb29f98a7a73f22da21ef85e0783f6fd67c4 Mon Sep 17 00:00:00 2001 From: Magnus Edenhill Date: Mon, 23 May 2016 19:58:05 +0200 Subject: [PATCH 4/4] Version 0.9.1 (release) --- src-cpp/rdkafkacpp.h | 2 +- src/rdkafka.h | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src-cpp/rdkafkacpp.h b/src-cpp/rdkafkacpp.h index ac0cdaed96..c9fdc3981a 100644 --- a/src-cpp/rdkafkacpp.h +++ b/src-cpp/rdkafkacpp.h @@ -87,7 +87,7 @@ namespace RdKafka { * @remark This value should only be used during compile time, * for runtime checks of version use RdKafka::version() */ -#define RD_KAFKA_VERSION 0x00090100 +#define RD_KAFKA_VERSION 0x000901ff /** * @brief Returns the librdkafka version as integer. diff --git a/src/rdkafka.h b/src/rdkafka.h index daccc04ca6..3d4a069f9e 100644 --- a/src/rdkafka.h +++ b/src/rdkafka.h @@ -98,7 +98,7 @@ typedef SSIZE_T ssize_t; * @remark This value should only be used during compile time, * for runtime checks of version use rd_kafka_version() */ -#define RD_KAFKA_VERSION 0x00090100 +#define RD_KAFKA_VERSION 0x000901ff /** * @brief Returns the librdkafka version as integer.