Librdkafka: Support for crc checking in the consumer

[HenningRoigaard]

Currently there is no crc checking when consuming, and the API does not expose enough information for a user of the library to perform the checking (most notably the crc of the message).

Preferably, the consumer should have built-in support for checking the crc, but alternately it would suffice to expose the information required for users of the library to perform the check.

As a reference, the java implementation checks crc by default and allows the user to opt out using the configuration “check.crcs” (see documentation).

This issue has previously been posted at Confluent.Kafka, as I experienced the issue using that library. See that issue for a description of how I tested the missing crc check using Confluent.Kafka.

[edenhill]

Now fixed on master, please give it a spin. (set check.crcs=true)

[HenningRoigaard]

@edenhill I would like to test it, but I am having difficulties building my master clone of librdkafka (head: caad14f) due to the errors listed below.
I have been following the instructions in the readme.win32 (incl. open-ssl install of specified version), but to no avail. My lack of experience building c/c++ projects might be to blame, and google did not provide any useful hints.
Are there any precompiled assemblies for download somewhere?

The build attempt was performed on a Windows 10 box using Visual Studio 2015.

1>------ Build started: Project: librdkafka, Configuration: Debug Win32 ------
1> Creating library C:\src\sibyl\librdkafka\win32\outdir\v140\Win32\Debug\librdkafka.lib and object C:\src\sibyl\librdkafka\win32\outdir\v140\Win32\Debug\librdkafka.exp
1>ssleay32MT.lib(d1_both.obj) : error LNK2001: unresolved external symbol ___iob_func
1>ssleay32MT.lib(s3_srvr.obj) : error LNK2001: unresolved external symbol ___iob_func
1>libeay32MT.lib(txt_db.obj) : error LNK2001: unresolved external symbol ___iob_func
1>libeay32MT.lib(cryptlib.obj) : error LNK2001: unresolved external symbol ___iob_func
1>libeay32MT.lib(pem_lib.obj) : error LNK2001: unresolved external symbol ___iob_func
1>libeay32MT.lib(ui_openssl.obj) : error LNK2001: unresolved external symbol ___iob_func
1>ssleay32MT.lib(t1_enc.obj) : error LNK2001: unresolved external symbol ___iob_func
1>libeay32MT.lib(cryptlib.obj) : error LNK2019: unresolved external symbol _sscanf referenced in function _OPENSSL_cpuid_setup
1>libeay32MT.lib(v3_utl.obj) : error LNK2001: unresolved external symbol _sscanf
1>libeay32MT.lib(cryptlib.obj) : error LNK2019: unresolved external symbol __vsnprintf referenced in function _OPENSSL_showfatal
1>libeay32MT.lib(dso_win32.obj) : error LNK2019: unresolved external symbol _sprintf referenced in function _win32_name_converter
1>C:\src\sibyl\librdkafka\win32\outdir\v140\Win32\Debug\librdkafka.dll : fatal error LNK1120: 4 unresolved externals
========== Build: 0 succeeded, 1 failed, 0 up-to-date, 0 skipped ==========

[edenhill]

librdkafka was changed for the v0.9.4 release to link OpenSSL statically, and since the OpenSSL we use is built with the v120 toolchain that means librdkafka is no longer buildable on VS 2015 (v140).

The quick fix is to go into win32/librdkafka.vcxproj and change any references to C:\OpenSSL-Win...\lib\VC\static to just ....\lib\VC.

I'm looking into a proper solution for this

[HenningRoigaard]

@edenhill I am afraid I have hit another snag. The project now builds, but it dawned on me, that I cannot (or do not know how to) test the library directly.
Hoping for the best, I tried to use Confluent.Kafka as an intermediary using the approach detailed below, but it failed.

Any ideas how to test the new feature using Confluent.Kafka, if at all possible? Or suggestions on another method altogether? If not, I fear I won't be able to test it for you. Sorry.

I tried using the following approach:

1. Create a new console application in visual studio 2015
2. Install Confluent.Kafka 0.9.4 via nuget
3. Build
4. Override the nuget librdkafka.dll with the locally build in the console applications Bin\Debug\x64 (I am running x64)
5. Run the following code:

static void Main(string[] args)
{
  var config = new Dictionary<string, object>()
  {
    { "enable.auto.commit", false },
    { "group.id", "NOT_A_GROUP" },
    { "bootstrap.servers", "localhost:9092" },
    { "check.crcs", true }
  };

  var consumer = new Consumer(config)
}

I then get an exception from the initializer of the consumer:

The type initializer for 'Confluent.Kafka.Impl.LibRdKafka' threw an exception.
at Confluent.Kafka.Impl.LibRdKafka.conf_new()
at Confluent.Kafka.Impl.SafeConfigHandle.Create()
at Confluent.Kafka.Consumer..ctor(IEnumerable`1 config)
at ConfluentKafkaCustomLibrdkafka.Program.Main(String[] args) in C:\src\sletmig\ConfluentKafkaCustomLibrdkafka\Program.cs:line 22
at System.AppDomain._nExecuteAssembly(RuntimeAssembly assembly, String[] args)
at System.AppDomain.ExecuteAssembly(String assemblyFile, Evidence assemblySecurity, String[] args)
at Microsoft.VisualStudio.HostingProcess.HostProc.RunUsersAssembly()
at System.Threading.ThreadHelper.ThreadStart_Context(Object state)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.ThreadHelper.ThreadStart()

with inner exception:

Unable to load DLL 'librdkafka': The specified module could not be found. (Exception from HRESULT: 0x8007007E)
at Confluent.Kafka.Impl.LibRdKafka.NativeMethods.rd_kafka_version()
at Confluent.Kafka.Impl.LibRdKafka.version()
at Confluent.Kafka.Impl.LibRdKafka..cctor()

[edenhill]

If you replace the new dll with the old librdkafka.dll it works again?
If so there is probably some kind of dependency problem.
You can verify the librdkafka.dll dependencies by running dumpbin /dependents librdkafka.dll and comparing that between the two versions.

Alternatively you could grab a dll from here which should be built the same way as the nuget dll.

[HenningRoigaard]

I made the mistake of doing a debug-build, which introduced additional dependencies. Changing to release-build solved the problem, and I am now able to do the test.
However setting check.crcs=true does not seem to do anything, but perhaps I do not know what to look for.

Before i describe my test-setup, what is the expected effect of check.crc=true when encountering a corrupt message? I expected an exception from Consumer.Consume, an error on the out'ed message or Consumer.OnError output.

[edenhill]

It should propagate CRC errors through Message->error() (as returned from consume() or poll()) with error code ERR__BAD_MSG.

Are you sure the CRC is actually incorrect?

Also make sure to use latest master, there was an error in the first commit for CRC checking, at least this commit or later: https://github.com/edenhill/librdkafka/commit/d116281d758bb739e9ab21f016adef819fe564ac

[HenningRoigaard]

The tip on my master is caad14f, which includes d116281.

How do i know if the CRC is incorrect
I have tested the missing crc checking by producing a simple message (key “123” and value “abc”), editing the log with visual studios binary editor to have key “143”:

As shown in the image above, the last value in the middle row has been changed from 32 to 34.
Consuming the message yielded a message with key “143” and no error. As the crc is not exposed when fetching a message in Confluent.Kafka, I could not directly check that the crc of the message was changed. To confirm that the crc was indeed changed, I wrote a simple client library using the Kafka Fetch API and calculated the crc using Crc32Provider from the kafka-net library. This confirmed that the crc was changed when flipping “123” to “143” and unchanged when not.
Perhaps the implementation i used to calculate the CRC32 is flawed?

How did i test check.crcs
The above mentioned message (key “123” and value “abc”) is produced to single-partition topic "CorruptionsTest" and corrupted.

I then did the following:

1. Create a new console application in visual studio 2015
2. Install Confluent.Kafka 0.9.4 via nuget
3. Build
4. Override the nuget librdkafka.dll and zlib.dll with the locally build in the console applications Bin\Debug\x86
5. Run the following code:

 static void Main(string[] args)
 {
   var config = new Dictionary<string, object>()
   {
     { "enable.auto.commit", false },
     { "group.id", "NOT_A_GROUP" },
     { "bootstrap.servers", "localhost:9092" },
     { "check.crcs", true }
   };

   using (var consumer = new Consumer(config))
   {
     consumer.OnError += (s,e) => Console.WriteLine(e);

     consumer.Assign(new List<TopicPartitionOffset>() { new TopicPartitionOffset("CorruptionsTest", 0, new Offset(0)) });

     Message message;
     var success = consumer.Consume(out message);

     if (success)
     {
       var key = Encoding.UTF8.GetString(message.Key);
       var value = Encoding.UTF8.GetString(message.Value);
       Console.WriteLine(key);
       Console.WriteLine(value);
     }
   }

   Console.WriteLine("Done - press key to end");
   Console.ReadKey();
}

Consumer.Consume returns true, the message has ErrorCode = NoError and the key "143" and value "abc" is printed to the console.