{"action":"opened","issue":{"url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/476","repository_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet","labels_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/476/labels{/name}","comments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/476/comments","events_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/476/events","html_url":"https://github.com/confluentinc/confluent-kafka-dotnet/issues/476","id":308149362,"number":476,"title":"Producer backpressure semantic is not async-friendly","user":{"login":"vchekan","id":106474,"avatar_url":"https://avatars1.githubusercontent.com/u/106474?v=4","gravatar_id":"","url":"https://api.github.com/users/vchekan","html_url":"https://github.com/vchekan","followers_url":"https://api.github.com/users/vchekan/followers","following_url":"https://api.github.com/users/vchekan/following{/other_user}","gists_url":"https://api.github.com/users/vchekan/gists{/gist_id}","starred_url":"https://api.github.com/users/vchekan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vchekan/subscriptions","organizations_url":"https://api.github.com/users/vchekan/orgs","repos_url":"https://api.github.com/users/vchekan/repos","events_url":"https://api.github.com/users/vchekan/events{/privacy}","received_events_url":"https://api.github.com/users/vchekan/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2018-03-23T19:15:39Z","updated_at":"2018-03-23T19:15:39Z","closed_at":null,"author_association":"NONE","body":"Description\r\n===========\r\nI am thinking how to implement producer correctly in high load async environment. Producer has 2 main modes, callback based (IDeliveryHandler) and Task based. Both have blockIfQueueFull parameter.\r\n\r\nLets start with task mode. If I do the following:\r\n```\r\nforeach (var n in batch) {\r\n    var message = await producer.ProduceAsync(topic, null, n, true).\r\n        ContinueWith(t => Interlocked.Increment(ref counter));\r\n}\r\n```\r\nthen I effectively produce one message per batch, which is no go. So, I need to accumulate messages and await on batch, which leads to something like this:\r\n```\r\nvar stream = Enumerable.Range(1, batchSize).Select(i => i.ToString());\r\nvar tasks = stream.Select(async msg => await producer.ProduceAsync(topic, null, msg, true).\r\n       ContinueWith(t => Interlocked.Increment(ref counter)));\r\nawait Task.WhenAll(tasks);\r\n```\r\nBut this is problematic too. If I have an infinite stream of source messages, then I do not know when my \"tasks\" will end (never), so i'll have to do linq batching. \r\n\r\nMy another worry is that `produceAsync` will block, but I do not quite understand how this will interplay with `await` portion of `await producer.ProduceAsync`. Are there any chances that Task scheduler will see blocking task, and will schedule new and new threads because from Task Scheduler perspective it is just long running task and it will try to schedule as many parrallel tasks as possible without being aware that this blocked task is actually a backpressure? Would it lead to task exhaust situation?\r\n\r\nBottom line is, i do not see, how use Task based API because we need to await for 2 different type of async flows: backpressuire and task delivery. And sync and async do not mix well. I have intuition that librdkafka blocking call should be converted to a task which can be awaited for. Does this sound reasonable?\r\n\r\nLet's consider delivery handler option. It is more appealing to me beacuse generating a task per message and then awaiting for every one of them sounds sub-optimal to me. Start with this implementation:\r\n```\r\n        class DeliveryHandler : IDeliveryHandler<Null, String>\r\n        {\r\n            public int counter = 0;\r\n            public bool MarshalData => true;\r\n\r\n            public void HandleDeliveryReport(Message<Null, string> message)\r\n            {\r\n                Interlocked.Increment(ref counter);\r\n            }\r\n        }\r\n\r\n........\r\n            var deliveryHandler = new DeliveryHandler();\r\n            Task.Run(() =>\r\n            {\r\n                using (var producer = new Producer<Null, string>(config, null, new StringSerializer(Encoding.UTF8)))\r\n                {\r\n                    var stream = Enumerable.Range(1, batchSize).Select(i => i.ToString());\r\n                    foreach (var msg in stream)\r\n                    {\r\n                        producer.ProduceAsync(topic, null, msg, true, deliveryHandler);\r\n                    }\r\n                }\r\n            }).Wait();\r\n\r\n```\r\nNow producing is synchrounous and batch overflow blocking call is propagated to the caller naturally. But there are problems: if you are operating in async-heavy environment, you still want buffer overflow to be exposed as awaitable entity, so you have to make more work to convert it to async.  And second problem is, that now I have no way of knowing about tasks failure. How do I handle failed tasks in Delivery Handler?\r\n\r\nChecklist\r\n=========\r\nPlease provide the following information:\r\n\r\n - [x] Confluent.Kafka nuget version: 0.11.3\r\n - [ ] Apache Kafka version: \r\n - [ ] Client configuration:\r\n - [ ] Operating system:\r\n - [ ] Provide logs (with \"debug\" : \"...\" as necessary in configuration)\r\n - [ ] Provide broker log excerpts\r\n - [ ] Critical issue\r\n"}}
{"actor":{"display_login":"vchekan"}}