Convert Kafka Thread related code example into Unity3d

Question

I have code snippet of Kafka Consumer which i have developed with the help of this and here it is:

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

            //CancellationTokenSource cts = new CancellationTokenSource();
            //var consumeTask = Task.Run(() =>
            //{
            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();//This line getting hange in unity
                            Avro.Field f;

                            //consumeResult.Value
                            consumeResult.Value.Schema.TryGetField("favorite_number", out f);
                            Console.WriteLine(consumeResult.Value["favorite_number"]);
                            Console.WriteLine(consumeResult.Value["name"]);

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

This is example is running fine in my console application. Now i want to move this example into Unity environment, for this reason i have make these changes. 1. Wrap the code inside coroutine, so it simultanously run with main unity thread. 2. Comment out catch block as i can't use them yeild return

 void Start()
    {
        StartCoroutine(Main());

    }

    IEnumerator Main()
    {

        string bootstrapServers = "localhost:9092";
        string schemaRegistryUrl = "Production163:8081";
        string topicName = "player";
        string groupName = "avro-generic-example-group";

        //CancellationTokenSource cts = new CancellationTokenSource();
        //var consumeTask = Task.Run(() =>
        //{
        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())
        {
            Debug.Log("subscribe");
            consumer.Subscribe(topicName);

            //try
            //{
            while (true)
            {
                //try
                //{
                //CancellationTokenSource cts = new CancellationTokenSource();
                //cts.Token
                if (consumer == null)
                {
                    Debug.Log("Consumer is null");
                }
                var consumeResult = consumer.Consume();//TimeSpan.FromMilliseconds(50000)
                yield return consumeResult;
                Debug.Log($"Key: {consumeResult.Message.Key}\nValue: {consumeResult.Value}");
                Debug.Log(consumeResult.Value.Schema);
                Debug.Log(consumeResult.Value.Schema["favorite_number"].GetProperty("favorite_number"));
                Debug.Log(consumeResult.Value.Schema["favorite_number"]);

                Avro.Field f;

                //consumeResult.Value

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

                Debug.Log(consumeResult.Value["favorite_number"]);
                Debug.Log(consumeResult.Value["name"]);

                //consumeResult.Message.Value.TryGetValue("favorite_number");

                //Debug.Log(f.);
                yield return new WaitForSeconds(1);

                //}
                //catch (ConsumeException e)
                //{
                //    Debug.Log($"Consume error: {e.Error.Reason}");
                //}

            }
            //}
            //catch (OperationCanceledException)
            //{
            //    // commit final offsets and leave the group.
            //    consumer.Close();
            //}
        }
        //});
        Debug.Log("Main end game.");
    }

Now with above code is my unity player get hanged. I try to debug and found that the problem is, in this line

var consumeResult = consumer.Consume();

The documentation suggest that there are two variants. One with TimeSpan and other with CancellationToken

Poll for new messages / events. Blocks until a consume result is available or the operation has been cancelled.

Answer 1

Wrap the code inside coroutine, so it simultanously run with main unity thread

You have fundamentally misunderstood how coroutines work. Consulting the Unity documentation:

A coroutine is like a function that has the ability to pause execution and return control to Unity but then to continue where it left off on the following frame.

(Emphasis added)

So: coroutines are not threads. They do not run simultaneously with the main thread. They take turns.

Each frame, Unity's main thread will iterate over each Coroutine that's set to resume at that time (based on its last yield return value), and call MoveNext() to run it until it hits its next yield return, yield break, or the end of the function. Only then does control return back to Unity to pump the next coroutine or continue with updating the game state and rendering the frame.

So, if you want to use threads to run in parallel (and you're not targeting WebGL), then use threads.

You can set up a producer-consumer queue, where your Kafka Consumer thread acts as the producer, stuffing results into a queue that scripts on your main thread can then extract and work with. It might look a bit like this:

public class MultithreadedExample : MonoBehaviour {

   bool _keepThreadRunning = true;
   Thread _thread;
   CancellationTokenSource _cancel;

   // The Concurrent version is important here, since we're  
   // accessing each end of this queue from a different thread.
   ConcurrentQueue<string> _stringsReceived;

   void Start() {
       // Prepare to cancel the operation if we need to.
       _cancel = new CancellationTokenSource();

       // Spin up our thread.
       _thread = new Thread(KafkaReader);
       _thread.Start(_cancel.Token);
   }

   void OnDestroy() {
       // If we never started our thread, we're done.
       if(_thread == null)
           return;

       // Tell the thread to stop on its next loop.
       _keepThreadRunning = false;
       // Abort the current consume action.
       _cancel.Cancel();
       // Wait until the thread has completed.
       _thread.Join();

       // Here the thread has safely exited, and we can clean up anything else we need.
   }

   void Update() {
       // Print out all messages we received since last frame.
       string message;
       while(_receivedStrings.TryDequeue(out message)) {
           Debug.LogFormat("{0} - received message: {1}", Time.time, message);
       }
   }


   void KafkaReader(CancellationToken cancellationToken) {

       // Set up your Kafka connection here.

       while(_keepThreadRunning) {
           try {
           var consumeResult = consumer.Consume(cancellationToken);

           // Do stuff with the values you've consumed.
           _receivedStrings.Enqueue(consumeResult.Value["favorite_number"]);

           // Handle or log errors...
           } catch (ConsumeException e) {
               _receivedStrings.Enqueue("Consume error: " + e.Error.Reason);
           }
       }

       // Disconnect and clean up your connection here.
   }    
}

If you need to target a platform that does not support threads, like Unity's WebGL target, then your best bet is to use the TimeSpan version of Consume(), with a short enough timespan that waiting on it will not disrupt your framerate.

Answer 2

Use the timeout alternative, with a zero second timeout, and wait for it to return non-null:

Turn this

var consumeResult = consumer.Consume();
yield return consumeResult;

Into this

ConsumeResult<string, GenericRecord> consumeResult;
do {
  consumeResult = consumer.Consume(TimeSpan.Zero);
  yield return consumeResult;
while (consumeResult == null);