1
\$\begingroup\$

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.

\$\endgroup\$

2 Answers 2

Reset to default
2
\$\begingroup\$

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.

\$\endgroup\$
4
  • \$\begingroup\$ Thanks for the answer, I will update the solution. But the given code should not be feasible for webgl so what kind of change will be require? \$\endgroup\$ Commented May 28, 2019 at 11:55
  • \$\begingroup\$ Look at the last paragraph where it says "If you need to target a platform that does not support threads, like Unity's WebGL target, then..." \$\endgroup\$ Commented May 28, 2019 at 11:55
  • \$\begingroup\$ Yeah i have read it but my original question with same parameter. \$\endgroup\$ Commented May 28, 2019 at 12:00
  • \$\begingroup\$ I have given timespan with 1 second but it get hang \$\endgroup\$ Commented May 28, 2019 at 12:01
0
\$\begingroup\$

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);
\$\endgroup\$

You must log in to answer this question.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.