Kafka in Unity not working. Seems to have broken assemblies or conflicting with Unity's. How to fix?

[Phantomb]

Description

I want to integrate Kafka in Unity for a Windows build (I am on Windows).

However I either get errors such as this one:

Error: Could not load signature of Confluent.Kafka.IAsyncDeserializer`1[T]:DeserializeAsync due to: Could not load file or assembly 'System.Memory, Version=4.0.1.0, Culture=neutral, PublicKeyToken=cc7b13ffcd2ddd51' or one of its dependencies. assembly:System.Memory, Version=4.0.1.0,

Or if I install the System.Memory version that comes with the Kafka NuGet dependency, I get errors all over the place of Unity's own method/class references no longer being found.

How to reproduce

I have tried installing the latest release (1.3.0) from nuget (using the UnityPackage 'NuGetForUnity'). This did not seem to work and had the second type of error result described above.

I have then tried to build the source and copy the relevant DLL's manually, and tried the same with the nuget release version. This gives the first type of error result described above.

Can anyone please help me with installing Kafka for Unity?

[mhowlett]

thanks for opening the issue @Phantomb . i would guess it shouldn't require too much effort to make the library compatible with unity, we haven't tested it.

[Phantomb]

i would guess it shouldn't require too much effort to make the library compatible with unity,

Could you tell me what should be done to make it compatible?

Also I see past issues opened by other people who implemented the library in Unity (and had different issues, I suppose). How did they go about this?

[tamirh]

What Unity version are you using? I tried quickly on a 2018.4 and 2019.1 with NuGetForUnity on a clean project and it imported fine without error. Do you get the compile errors in editor, or only on a build?

If you go to your PlayerSettings, what do you have for:

• Scripting Runtime Version
• Scripting Backend
• API Compatibility Level
• Enable Unsafe Code

Try enabling the "Unsafe Code" checkbox, since that will allow Unity to import System.Memory which it doesn't by default. You may have to remove the duplicate packages that NuGetForUnity downloaded that duplicate the .net ones.

[Phantomb]

Thanks for the reply @tamirh

I'm tried it on Unity 2019.2.0f1 in an existing project. I get the compile errors in the editor.

These are my Player settings:

Scripting Runtime Version is no longer present on 2019.2 it seems.

Enabling "Unsafe Code" does not seem to help at my end.

[Phantomb]

@tamirh when testing in a fresh Unity project, importing using NuGetForUnity seems to work. However, after I add the basic consumer (from the main readme of this repo), it says it fails to build (see image and code below).

Do you have any idea how to fix this?

I have kafka and zookeeper server processes running. The basis producer example I copied seems to work (the Produce method version), and that producer shows up in the server info prints.

using Confluent.Kafka;
using System;
using System.Threading;
using UnityEngine;

class KafkaConsumer: MonoBehaviour
{
    void Start()
    {
        Debug.Log("Start");
        var config = new ConsumerConfig{
                GroupId = "c#test-consumer-group"+DateTime.Now,  // unique group, so each listener gets all messages
                BootstrapServers = "127.0.0.1:9092",
                AutoOffsetReset = AutoOffsetReset.Latest
            };
        main(config);
    }
    void main(ConsumerConfig config)
    {
        Debug.Log("Build Kafka Consumer");
        using (var c = new ConsumerBuilder<Ignore, string>(config).Build())
        {
            c.Subscribe("my-topic");
            Debug.Log("Subscribed kafka");

[Phantomb]

I also tried the 1.4.0-RC2 but same result.

[Phantomb]

Just to update: managed to fix this a while ago by both changing which lib files I had (where), and by changing the consumer code. Thanks for the replies, closing this issue.

Image of current set up of libs & scripts:

Kafka consumer code to support Unity:

using Confluent.Kafka;
using System;
using System.Threading;
using UnityEngine;
using System.Collections.Concurrent;

class KafkaConsumer: MonoBehaviour
{
    
    [System.Serializable]
    public class threadHandle
    {
        // IConsumer<Ignore, string> c;
        ConsumerConfig config;

        public readonly ConcurrentQueue<StreamMessage> _queue = new ConcurrentQueue<StreamMessage>();
        public void StartKafkaListener()
        {
            Debug.Log("Kafka - Starting Thread..");
            try 
            {
                config = new ConsumerConfig{
                        GroupId = "c#test-consumer-group"+DateTime.Now,  // unique group, so each listener gets all messages
                        BootstrapServers = "localhost:9092",
                        AutoOffsetReset = AutoOffsetReset.Earliest
                    };

                Debug.Log("Kafka - Created config");

                using (var c = new ConsumerBuilder<Ignore, byte[]>(config).Build())
                        {
                            c.Subscribe("networktopic_data");
                            Debug.Log("Kafka - Subscribed");

                            CancellationTokenSource cts = new CancellationTokenSource();
                            Console.CancelKeyPress += (_, e) => {
                                e.Cancel = true; // prevent the process from terminating.
                                cts.Cancel();
                            };

                            try
                            {
                                while (true)
                                {
                                    try
                                    {
                                        // Waiting for message
                                        var cr = c.Consume(cts.Token);
                                        // Got message! Decode and put on queue
                                        StreamMessage message = ParseStreamMessage.Decode(cr.Value);
                                        _queue.Enqueue(message);
                                    }
                                    catch (ConsumeException e)
                                    {
                                        Debug.Log("Kafka - Error occured: " + e.Error.Reason);
                                    }
                                }
                            }
                            catch (OperationCanceledException)
                            {
                                Debug.Log("Kafka - Canceled..");
                                // Ensure the consumer leaves the group cleanly and final offsets are committed.
                                c.Close();
                            }
                        }             
            }
            catch (Exception ex) {
                Debug.Log("Kafka - Received Expection: " + ex.Message + " trace: " + ex.StackTrace);
            }
        }
    }

    bool kafkaStarted = false;
    Thread kafkaThread;
    threadHandle _handle;

    void Start()
    {
        StartKafkaThread();
    }
    void Update()
    {
        if (Input.GetKeyUp(KeyCode.LeftControl) && Input.GetKeyUp(KeyCode.C))
        {
            Debug.Log("Cancelling Kafka!");
            StopKafkaThread();
        }

        ProcessKafkaMessage();
    }

    void OnDisable()
    {
        StopKafkaThread();        
    }
    void OnApplicationQuit()
    {
        StopKafkaThread();
    }

    public void StartKafkaThread()
    {
        if(kafkaStarted) return;

        _handle = new threadHandle();
        kafkaThread = new Thread(_handle.StartKafkaListener);

        kafkaThread.Start();
        kafkaStarted = true;
        // StartKafkaListener(config);
    }
    private void ProcessKafkaMessage()
    {
        if (kafkaStarted)
        {
            StreamMessage message;
            while (_handle._queue.TryDequeue(out message))
            {
                ProcessPBMessages.Process(message);
            }
        }
    }

    void StopKafkaThread()
    {
        if(kafkaStarted)
        {
            kafkaThread.Abort();
            kafkaThread.Join();
            kafkaStarted = false;
        }
    }
}

[ChulgooKim]

What is the source of the StreamMessage ?

[Phantomb]

@ChulgooKim StreamMessage is just a struct I used to process / structure the sent data and was not relevant to the issue (which I fixed, see my latest comment.

[ztolley]

@Phantomb How did you figure that out? I'm relatively new to Unity and I'm trying to do a simple proof of concept.

I added the Kafka dotnet package into my project and I get the same error you describe but don't know how to fix it.

[Mason-McGough]

@ztolley I needed to do the same thing and recently figured it out. You may be missing the librdkafka.dll file for your system. For reference, here are the steps I used to get Kafka working in my project:

1. Download and install the NuGet for Unity .unitypackage from their GitHub.
2. In Unity, open "NuGet"->"Manage NuGet Packages" and install "Confluent.Kafka".
3. From the NuGet package page for librdkafka, click "Download Package".
4. Extract the archive. Under the "runtimes" folder, copy the folder for your system (I am on Ubuntu 18.04 so I picked "linux-x64") and paste it under "Assets/Packages/Confluent.Kafka./lib/netstandard2.0" in Unity.
5. Create a C# script called "KafkaConsumer" and paste the code @Phantomb provided.
6. If needed, change BootstrapServers = "localhost:9092" to reflect your server and c.Subscribe("networktopic_data"); to the name of the topic you wish to subscribe to.
7. You might need to modify a few more things to get it to work for your case. To get started, find and replace StreamMessage with string for now. Replace StreamMessage message = ParseStreamMessage.Decode(cr.Value); with string message = Encoding.UTF8.GetString(cr.Value);. Replace ProcessPBMessages.Process(message); with Debug.Log(message);. Once you can play the scene and confirm you are receiving messages in the Console, you can try using Newtonsoft JsonConvert or something similar to work for your use case.

Hope that helps!

[KyleNeubarth]

Using @Mason-McGough's answer I'm still getting the same issue. it's a great step by step process though. I was missing librdkafka.dll in my Unity project, so I assumed that was the issue but even with the dll in the correct folder I didn't get a difference.

Here's the traceback using @Phantomb's code:

Kafka - Received Expection: Failed to load the librdkafka native library. trace:   at Confluent.Kafka.Impl.Librdkafka.TrySetDelegates (System.Collections.Generic.List`1[T] nativeMethodCandidateTypes) [0x00034] in <f055ff2284d34a38b90c6896d0be9574>:0 
  at Confluent.Kafka.Impl.Librdkafka.LoadNetStandardDelegates (System.String userSpecifiedPath) [0x0004b] in <f055ff2284d34a38b90c6896d0be9574>:0 
  at Confluent.Kafka.Impl.Librdkafka.Initialize (System.String userSpecifiedPath) [0x0002a] in <f055ff2284d34a38b90c6896d0be9574>:0 
  at Confluent.Kafka.Consumer`2[TKey,TValue]..ctor (Confluent.Kafka.ConsumerBuilder`2[TKey,TValue] builder) [0x00169] in <f055ff2284d34a38b90c6896d0be9574>:0 
  at Confluent.Kafka.ConsumerBuilder`2[TKey,TValue].Build () [0x00000] in <f055ff2284d34a38b90c6896d0be9574>:0 
  at KafkaConsumer+threadHandle.StartKafkaListener () [0x00053] in C:\Users\Kyle N\Projects\Testing\KafkaTemp\Assets\KafkaConsumer.cs:32 
UnityEngine.Debug:Log (object)
KafkaConsumer/threadHandle:StartKafkaListener () (at Assets/KafkaConsumer.cs:71)
System.Threading.ThreadHelper:ThreadStart ()

The main mystery is why the librdkafka.redist dependency's dll doesn't show up in Unity. However if you check in the .nuget folder under the user, it's all there! Was hoping that getting Kafka running in Unity would have a simpler solution but I might try and compile the library using Visual Studio to see if that works.