Node-rdkafka: Int64 values in binary key get corrupted at consume.

Created on 30 Jun 2017 · 16Comments · Source: Blizzard/node-rdkafka

We use thrift encoded binary data as key for our kafka messages. When using node-rdkafka consumers the Int64 values seem to be getting corrupted in these binary keys. I can send a few samples:

Expected value: [Int64 value:132 octets:00 00 00 00 00 00 00 84]
Received value: [Int64 value:253 octets:00 00 00 00 00 00 00 fd]

Expected value: [Int64 value:325235 octets:00 00 00 00 00 04 f6 73]
Received value: [Int64 value:327027 octets:00 00 00 00 00 04 fd 73]

Expected value: [Int64 value:23444 octets:00 00 00 00 00 00 5b 94]
Received value: [Int64 value:23549 octets:00 00 00 00 00 00 5b fd]

Expected value: [Int64 value:23434 octets:00 00 00 00 00 00 5b 8a]
Received value: [Int64 value:23549 octets:00 00 00 00 00 00 5b fd]

The octets seem to be overflowing to 'fd' for some reason. Does anyone know a reason for this, or if the key of the message gets changed at any point in the rdkafka lib or in node at consumption?

I can create a sample application with docker-compose if that helps.

Note: When I use kafka-node I dont have this issue.
I tried producing this messages with both kafka-node and node-rdkafka. In both cases the kafka-node received the data correctly, while node-rdkafka got the Int64 values corrupted.

implemented

Source

utizr

👍1

All 16 comments

I found out what the issue is. The key is turned in to a string without specifying the source encoding, which in my case is binary. The toString() of node is using 'utf-8' by default.
I believe it is this part of the callback.cc in the src/.
Nan::New<v8::String>(event.key).ToLocalChecked());

Can anyone suggest me how to return the key as a buffer instead of a string? Other solution that would work is to be able to specify the toString encoding source at that point to 'binary'.
Then I can make a fork of the lib that we can use in our app.

utizr on 4 Jul 2017

I believe the 'fd' that i see in the octets is coming from the utf-8 replacement character :
unicode replacement character
So the issue is very likely with the string conversion of the key, because it is assumed the key encoding is utf-8.

utizr on 4 Jul 2017

Thanks for the sleuthing! I would like to make this so keys get returned as buffers just like values do. I believe that would solve your problem.

I'll try to get a build out for you test in the next few days and we'll see if that satisfies the requirements.

webmakersteve on 7 Jul 2017

👍1

@webmakersteve do you have any news about the buffered key version?

utizr on 7 Aug 2017

I'm trying to cut a new release but there is a potential regression in the new librdkafka version that is stopping me.

Changing keys to buffers is a breaking change so I need to make a major bump to make it work, and I'd like to batch it with the change to the librdkafka version and potentially one other breaking change before I do it to avoid bumping up another major.

webmakersteve on 9 Aug 2017

Same issue occurs with us. We have binary values on the message key, when parsing it, the buffers are corrupted somewhere in between kafka and our logic.

ruleb on 16 Aug 2017

Looks like a fix to the bug stopping the upgrade has been fixed here: https://github.com/edenhill/librdkafka/pull/1387

webmakersteve on 21 Aug 2017

👍1

https://github.com/Blizzard/node-rdkafka/pull/240

PR for buffer keys. Please test when you can, but I added an end to end test case as well.

webmakersteve on 22 Aug 2017

Thank you, I will check this later today!

utizr on 22 Aug 2017

@webmakersteve I am trying to test this with checking out the latest master commit in my application's package.json. The code is the latest when I check the node_modules, but I receive the following error at starting the app:

/usr/local/bin/node: symbol lookup error: /app/node_modules/node-rdkafka/build/Release/node-librdkafka.node: undefined symbol: rd_kafka_sasl_global_init

Do I need to do an extra step when cloning from master like pulling/building the rdkafka c lib?

utizr on 22 Aug 2017

There shouldn't be any extra steps as long as you are running npm install after you pull it down.

If you put it in package.json as sourcing from Github it should work as well. The error you're seeing suggests that SASL is installed, but it was linked properly.

Can you just run node-gyp rebuild when you're in the node-rdkafka directory and capture the output?

webmakersteve on 22 Aug 2017

@webmakersteve
This is the output of node-gyp rebuild

gyp info it worked if it ends with ok
gyp info using [email protected]
gyp info using [email protected] | linux | x64
gyp info spawn /usr/bin/python2
gyp info spawn args [ '/usr/local/lib/node_modules/npm/node_modules/node-gyp/gyp/gyp_main.py',
gyp info spawn args   'binding.gyp',
gyp info spawn args   '-f',
gyp info spawn args   'make',
gyp info spawn args   '-I',
gyp info spawn args   '/app/node_modules/node-rdkafka/build/config.gypi',
gyp info spawn args   '-I',
gyp info spawn args   '/usr/local/lib/node_modules/npm/node_modules/node-gyp/addon.gypi',
gyp info spawn args   '-I',
gyp info spawn args   '/root/.node-gyp/8.0.0/include/node/common.gypi',
gyp info spawn args   '-Dlibrary=shared_library',
gyp info spawn args   '-Dvisibility=default',
gyp info spawn args   '-Dnode_root_dir=/root/.node-gyp/8.0.0',
gyp info spawn args   '-Dnode_gyp_dir=/usr/local/lib/node_modules/npm/node_modules/node-gyp',
gyp info spawn args   '-Dnode_lib_file=node.lib',
gyp info spawn args   '-Dmodule_root_dir=/app/node_modules/node-rdkafka',
gyp info spawn args   '-Dnode_engine=v8',
gyp info spawn args   '--depth=.',
gyp info spawn args   '--no-parallel',
gyp info spawn args   '--generator-output',
gyp info spawn args   'build',
gyp info spawn args   '-Goutput_dir=.' ]
gyp info spawn make
gyp info spawn args [ 'BUILDTYPE=Release', '-C', 'build' ]
make: Entering directory '/app/node_modules/node-rdkafka/build'
  ACTION configuring librdkafka... deps/librdkafka/config.h
using cache file config.cache
checking for OS or distribution... ok (Linux)
checking for C compiler from CC env... failed
checking for gcc (by command)... ok (cached)
checking for C++ compiler from CXX env... failed
checking for C++ compiler (g++)... ok (cached)
checking executable ld... ok (cached)
checking executable nm... ok (cached)
checking executable objdump... ok (cached)
checking executable strip... ok (cached)
checking for pkgconfig (by command)... ok (cached)
checking for install (by command)... ok (cached)
checking for PIC (by compile)... ok (cached)
checking for GNU-compatible linker options... ok (cached)
checking for GNU linker-script ld flag... ok (cached)
checking for __atomic_32 (by compile)... ok (cached)
checking for __atomic_64 (by compile)... ok (cached)
checking for socket (by compile)... ok (cached)
parsing version '0x000b00ff'... ok (0.11.0)
checking for libpthread (by pkg-config)... failed
checking for libpthread (by compile)... ok (cached)
checking for zlib (by pkg-config)... ok
checking for zlib (by compile)... ok (cached)
checking for libcrypto (by pkg-config)... ok
checking for libcrypto (by compile)... ok (cached)
checking for liblz4 (by pkg-config)... failed
checking for liblz4 (by compile)... failed (disable)
checking for libssl (by pkg-config)... ok
checking for libssl (by compile)... ok (cached)
checking for crc32chw (by compile)... ok (cached)
checking for regex (by compile)... ok (cached)
checking for librt (by pkg-config)... failed
checking for librt (by compile)... ok (cached)
checking for strndup (by compile)... ok (cached)
checking for libdl (by pkg-config)... failed
checking for libdl (by compile)... ok (cached)
checking for nm (by env NM)... ok (cached)
checking for python (by command)... ok (cached)
Generated Makefile.config
Generated config.h

Configuration summary:
  prefix                   /usr/local
  ARCH                     x86_64
  CPU                      generic
  GEN_PKG_CONFIG           y
  ENABLE_DEVEL             n
  ENABLE_VALGRIND          n
  ENABLE_REFCNT_DEBUG      n
  ENABLE_SHAREDPTR_DEBUG   n
  ENABLE_LZ4_EXT           y
  ENABLE_SSL               y
  ENABLE_SASL              n
  MKL_APP_NAME             librdkafka
  MKL_APP_DESC_ONELINE     The Apache Kafka C/C++ library
  MKL_DISTRO               Linux
  SOLIB_EXT                .so
  CC                       gcc
  CXX                      g++
  LD                       ld
  NM                       nm
  OBJDUMP                  objdump
  STRIP                    strip
  CPPFLAGS                 -g -O2 -fPIC -Wall -Wsign-compare -Wfloat-equal -Wpointer-arith -Wcast-align
  PKG_CONFIG               pkg-config
  INSTALL                  install
  LIB_LDFLAGS              -shared -Wl,-soname,$(LIBFILENAME)
  LDFLAG_LINKERSCRIPT      -Wl,--version-script=
  RDKAFKA_VERSION_STR      0.11.0
  MKL_APP_VERSION          0.11.0
  LIBS                     -lpthread -L/usr/lib/x86_64-linux-gnu -lz  -lcrypto  -lssl -lcrypto  -lrt -ldl
  CFLAGS
  CXXFLAGS                 -Wno-non-virtual-dtor
  SYMDUMPER                $(NM) -D
  exec_prefix              /usr/local
  bindir                   /usr/local/bin
  sbindir                  /usr/local/sbin
  libexecdir               /usr/local/libexec
  datadir                  /usr/local/share
  sysconfdir               /usr/local/etc
  sharedstatedir           /usr/local/com
  localstatedir            /usr/local/var
  libdir                   /usr/local/lib
  includedir               /usr/local/include
  infodir                  /usr/local/info
  mandir                   /usr/local/man
Generated config.cache

Now type 'make' to build
  TOUCH Release/obj.target/deps/librdkafka_config.stamp
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_buf.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdcrc32.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/crc32c.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_metadata.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_transport.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_range_assignor.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_queue.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_offset.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/tinycthread.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_subscription.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_msgset_reader.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdrand.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_interceptor.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_lz4.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdunittest.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_broker.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_cgrp.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/lz4frame.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdvarint.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rddl.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_metadata_cache.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdlog.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/snappy.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_feature.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_request.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/lz4hc.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/regexp.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_msgset_writer.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdports.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_plugin.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_msg.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdstring.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdavl.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_conf.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_assignor.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdaddr.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_timer.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_partition.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_topic.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/lz4.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdgz.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_event.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdlist.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdbuf.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_op.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_pattern.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/xxhash.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdregex.o
  CC(target) Release/obj.target/librdkafka/deps/librdkafka/src/rdkafka_roundrobin_assignor.o
  AR(target) Release/obj.target/deps/rdkafka.a
  COPY Release/rdkafka.a
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/ConfImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/MetadataImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/ConsumerImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/TopicPartitionImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/MessageImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/QueueImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/HandleImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/KafkaConsumerImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/TopicImpl.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/RdKafka.o
  CXX(target) Release/obj.target/librdkafka_cpp/deps/librdkafka/src-cpp/ProducerImpl.o
  AR(target) Release/obj.target/deps/rdkafka_cpp.a
  COPY Release/rdkafka_cpp.a
  CXX(target) Release/obj.target/node-librdkafka/src/binding.o
  CXX(target) Release/obj.target/node-librdkafka/src/callbacks.o
  CXX(target) Release/obj.target/node-librdkafka/src/common.o
  CXX(target) Release/obj.target/node-librdkafka/src/config.o
  CXX(target) Release/obj.target/node-librdkafka/src/connection.o
  CXX(target) Release/obj.target/node-librdkafka/src/errors.o
  CXX(target) Release/obj.target/node-librdkafka/src/kafka-consumer.o
  CXX(target) Release/obj.target/node-librdkafka/src/producer.o
  CXX(target) Release/obj.target/node-librdkafka/src/topic.o
  CXX(target) Release/obj.target/node-librdkafka/src/workers.o
  SOLINK_MODULE(target) Release/obj.target/node-librdkafka.node
  COPY Release/node-librdkafka.node
make: Leaving directory '/app/node_modules/node-rdkafka/build'
gyp info ok
npm info lifecycle [email protected]~postrebuildrdkafka: [email protected]
npm info ok

utizr on 22 Aug 2017

ENABLE_SASL was set to n but node-rdkafka requires it. Can you install sasl on your machine and run it again and ensure ENABLE_SASL is set to y? I think that's the problem.

webmakersteve on 22 Aug 2017

It works, and the int value is returned correctly!! Thank you again @webmakersteve.
If anyone has above issue, I had the env variable WITH_SASL=0 set in my docker file, so that node-rdkafka can install without errors. This was also the reason for the not set ENABLE_SASL that webmakersteve mentioned. To fix the issue I had to remove this env variable, and install the following 2 packages (ubuntu docker) libsasl2-dev libsasl2-modules before npm installing node-rdkafka.

utizr on 22 Aug 2017

Awesome. Glad to hear it. I am going to try to make a release that rolls back the librdkafka upgrade to 0.11 so i can get this out earlier.

webmakersteve on 23 Aug 2017

👍1

This has been released with 2.0.0

webmakersteve on 29 Aug 2017

👍1

Was this page helpful?

0 / 5 - 0 ratings