Kafka Consumer integration in unity3d C#

Question

I have this code snippet in C# console application. It is connecting and working fine in console-app environment but when i move this code to Unity3d environment it is through null reference exception. Maybe Unity need different way to impelment this as unity is not thread safe while the original example is developed using threads.

Console APP Snippet

static void Main(string[] args)
{
    string bootstrapServers = "localhost:9092";
    string schemaRegistryUrl = "Production163:8081";
    string topicName = "player";
    string groupName = "avro-generic-example-group";

    using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
         using (var consumer =
            new 
            ConsumerBuilder<string, GenericRecord>(new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = groupName })
                 .SetKeyDeserializer(new AsyncAvroDeserializer<string>(schemaRegistry).AsSyncOverAsync())
                 .SetValueDeserializer(new AsyncAvroDeserializer<GenericRecord>(schemaRegistry).AsSyncOverAsync())
                .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
                 .Build())
         {
             consumer.Subscribe(topicName);

             try
             {
                while (true)
                {
                     try
                     {
                         var consumeResult = consumer.Consume();
                        Avro.Field f;
                        consumeResult.Value.Schema.TryGetField("favorite_number", out f);

                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Consume error: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // commit final offsets and leave the group.
                consumer.Close();
            }
        }
}

But I want to run the same code snippet in Unity3d, I have to make some change as unity is not thread safe so I write this code with changes.

Unity Example

IEnumerator Main()
{
    string bootstrapServers = "localhost:9092";
    string schemaRegistryUrl = "Production163:8081";
    string topicName = "player";
    string groupName = "avro-generic-example-group";

    using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
    using (
        var consumer = new ConsumerBuilder<string, GenericRecord>(new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = groupName })
            .SetKeyDeserializer(new AsyncAvroDeserializer<string>(schemaRegistry).AsSyncOverAsync())
            .SetValueDeserializer(new AsyncAvroDeserializer<GenericRecord>(schemaRegistry).AsSyncOverAsync())
            .SetErrorHandler((_, e) => Debug.Log($"Error: {e.Reason}"))
            .Build())
    {

        consumer.Subscribe(topicName);

        while (true)
        {
                var consumeResult = consumer.Consume(TimeSpan.FromMilliseconds(1000));//TimeSpan.FromMilliseconds(50000)//this line through the exception
            Avro.Field f;

            consumeResult.Value.Schema.TryGetField("favorite_number", out f);

            yield return new WaitForSeconds(1);

        }
    }
    Debug.Log("Main end game.");
}

But the code is executing and throws Null Reference Exception in this line

  var consumeResult = `consumer.Consume(TimeSpan.FromMilliseconds(1000));//TimeSpan.FromMilliseconds(50000)//this line through the exception`

Error

NullReferenceException: Object reference not set to an instance of an object KafkaAvroConsumer+d__1.MoveNext () (at Assets/KafkaUnityIntegration/Scripts/KafkaAvroConsumer.cs:51) UnityEngine.SetupCoroutine.InvokeMoveNext (System.Collections.IEnumerator enumerator, System.IntPtr returnValueAddress) (at C:/buildslave/unity/build/Runtime/Export/Coroutines.cs:17)

Remember, I am using the error code line (consumer.Consume())in console app without any parameter.

 var consumeResult = consumer.Consume();

Note: I know that null reference is one of the common error but i am unable to find that why it is occurring as the same code snippet working fine in console app environment.

and what about using a second thread and rather make it "thread save" by e.g. using an `Action` worker that tells the Unity main thread what to do when done? — derHugo, May 24 '19 at 12:09
e.g. https://github.com/PimDeWitte/UnityMainThreadDispatcher — derHugo, May 24 '19 at 12:19
Could you clean up the second code exampe (i.e. remove the commented out lines) and indicate the specific line that is throwing the exception? — Colin Young, May 24 '19 at 12:28
@ColinYoung Yes, I have updated the code and specify the error line. — Muhammad Faizan Khan, May 25 '19 at 05:10
@derHugo thanks for the link. Do you mean i can use the same example using the threaddispatcer code? — Muhammad Faizan Khan, May 25 '19 at 05:55
I don't know since I don't know what it does or how works or where the exception comes from exactly. I just commented on your statement `as unity is not thread safe` that you can still use threads in Unity. You just have to hand the results back to the main thread using a `Queue` as e.g. done in this dispatcher thingy — derHugo, May 25 '19 at 07:00
Sorry I never experience Queue before, more details are much appreciated. and i have added where i am getting null refrence it is here consumer.Consume(). — Muhammad Faizan Khan, May 25 '19 at 07:36
@derHugo I have tried your dispatcher link but it get hanged. — Muhammad Faizan Khan, May 25 '19 at 07:37
Have you tried without the `TimeSpan.FromMilliseconds(1000)` in the Consume method? Or try passing a `TimeSpan` in the console app. It appears as though that overload is fairly new, so there may be bugs. — Colin Young, May 26 '19 at 03:10

Kafka Consumer integration in unity3d C#

0 Answers0