{"action":"created","issue":{"url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/470","repository_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet","labels_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/470/labels{/name}","comments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/470/comments","events_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/470/events","html_url":"https://github.com/confluentinc/confluent-kafka-dotnet/issues/470","id":307051209,"number":470,"title":"Why is Commit required after reading message off a topic for Consumer","user":{"login":"PingPongSet","id":2021414,"avatar_url":"https://avatars0.githubusercontent.com/u/2021414?v=4","gravatar_id":"","url":"https://api.github.com/users/PingPongSet","html_url":"https://github.com/PingPongSet","followers_url":"https://api.github.com/users/PingPongSet/followers","following_url":"https://api.github.com/users/PingPongSet/following{/other_user}","gists_url":"https://api.github.com/users/PingPongSet/gists{/gist_id}","starred_url":"https://api.github.com/users/PingPongSet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/PingPongSet/subscriptions","organizations_url":"https://api.github.com/users/PingPongSet/orgs","repos_url":"https://api.github.com/users/PingPongSet/repos","events_url":"https://api.github.com/users/PingPongSet/events{/privacy}","received_events_url":"https://api.github.com/users/PingPongSet/received_events","type":"User","site_admin":false},"labels":[{"id":473160567,"url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/labels/question","name":"question","color":"cc317c","default":true}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":6,"created_at":"2018-03-20T21:48:25Z","updated_at":"2018-03-21T21:56:38Z","closed_at":null,"author_association":"NONE","body":"Description\r\n===========\r\nI want to understand why Commit() is required (either auto/manual commit), after reading messages off a topic (either using Poll() or Consume() ).\r\n\r\nBelow is the modified code via .NET using Polling combined with manual CommitAsync() based on the code. https://github.com/confluentinc/confluent-kafka-dotnet/blob/master/examples/AdvancedConsumer/Program.cs\r\n\r\n```\r\n  public static void Run_PollWithManualCommit(string brokerList, List<string> topics)\r\n        {\r\n            using (var consumer = new Consumer<Ignore, string>(constructConfig(brokerList, false), null, new StringDeserializer(Encoding.UTF8)))\r\n            {\r\n                // Note: All event handlers are called on the main thread.\r\n\r\n                consumer.OnMessage += (_, msg)\r\n                    =>\r\n                {\r\n                    Console.WriteLine($\"Topic: {msg.Topic} Partition: {msg.Partition} Offset: {msg.Offset} {msg.Value}\");\r\n                    Console.WriteLine($\"Committing offset\");\r\n                    var committedOffsets = consumer.CommitAsync(msg).Result;\r\n                    Console.WriteLine($\"Committed offset: [{string.Join(\", \", committedOffsets.Offsets)}]\");\r\n                };\r\n\r\n                consumer.OnPartitionEOF += (_, end)\r\n            => Console.WriteLine($\"Reached end of topic {end.Topic} partition {end.Partition}, next message will be at offset {end.Offset}\");\r\n\r\n                // Raised on critical errors, e.g. connection failures or all brokers down.\r\n                consumer.OnError += (_, error)\r\n                    => Console.WriteLine($\"Error: {error}\");\r\n\r\n                // Raised on deserialization errors or when a consumed message has an error != NoError.\r\n                consumer.OnConsumeError += (_, msg)\r\n                    => Console.WriteLine($\"Error consuming from topic/partition/offset {msg.Topic}/{msg.Partition}/{msg.Offset}: {msg.Error}\");\r\n\r\n                //this is NOT called, when autocommit is disabled\r\n                consumer.OnOffsetsCommitted += (_, commit) =>\r\n                {\r\n                    Console.WriteLine($\"[{string.Join(\", \", commit.Offsets)}]\");\r\n\r\n                    if (commit.Error)\r\n                    {\r\n                        Console.WriteLine($\"Failed to commit offsets: {commit.Error}\");\r\n                    }\r\n                    Console.WriteLine($\"Successfully committed offsets: [{string.Join(\", \", commit.Offsets)}]\");\r\n                };\r\n\r\n                consumer.OnPartitionsAssigned += (_, partitions) =>\r\n                {\r\n                    Console.WriteLine($\"Assigned partitions: [{string.Join(\", \", partitions)}], member id: {consumer.MemberId}\");\r\n                    consumer.Assign(partitions);\r\n                };\r\n\r\n                consumer.OnPartitionsRevoked += (_, partitions) =>\r\n                {\r\n                    Console.WriteLine($\"Revoked partitions: [{string.Join(\", \", partitions)}]\");\r\n                    consumer.Unassign();\r\n                };\r\n\r\n                //consumer.OnStatistics += (_, json)\r\n                //    => Console.WriteLine($\"Statistics: {json}\");\r\n\r\n                //The subscribe() method controls which topics will be fetched in poll.\r\n                consumer.Subscribe(topics);\r\n\r\n                Console.WriteLine($\"Subscribed to: [{string.Join(\", \", consumer.Subscription)}]\");\r\n\r\n                var cancelled = false;\r\n                Console.CancelKeyPress += (_, e) => {\r\n                    e.Cancel = true; // prevent the process from terminating.\r\n                    cancelled = true;\r\n                };\r\n\r\n                Console.WriteLine(\"Ctrl-C to exit.\");\r\n                while (!cancelled)\r\n                {\r\n                    consumer.Poll(TimeSpan.FromMilliseconds(100));\r\n                }\r\n            }\r\n        }\r\n\r\n       private static Dictionary<string, object> constructConfig(string brokerList, bool enableAutoCommit) =>\r\n            new Dictionary<string, object>\r\n            {\r\n                { \"group.id\", \"advanced-csharp-consumer\" },\r\n                { \"enable.auto.commit\", enableAutoCommit },\r\n                { \"auto.commit.interval.ms\", 5000 },\r\n                { \"statistics.interval.ms\", 60000 },\r\n                { \"bootstrap.servers\", brokerList },\r\n                { \"default.topic.config\", new Dictionary<string, object>()\r\n                    {\r\n                        { \"auto.offset.reset\", \"smallest\" }\r\n                    }\r\n                }\r\n            };\r\n```\r\n\r\nI disabled the lines like below, and run Consumer again, it can still remember the CORRECT offset. That is,correct offset is used using Poll() method with manual commit without calling CommitAsync().\r\n\r\n          consumer.OnMessage += (_, msg)\r\n                    =>\r\n                {\r\n                    Console.WriteLine($\"Topic: {msg.Topic} Partition: {msg.Partition} Offset: {msg.Offset} {msg.Value}\");\r\n                    //Console.WriteLine($\"Committing offset\");\r\n                   // var committedOffsets = consumer.CommitAsync(msg).Result;\r\n                   // Console.WriteLine($\"Committed offset: [{string.Join(\", \", committedOffsets.Offsets)}]\");\r\n                };\r\n\r\nFor example, below is the test\r\n\r\nStep 1 Run Consumer to consume one message, \r\nStep 2 kill it\r\nStep 3 re-start it\r\nStep 4 it can still remember the correct offset each time after it re-start like below\r\n```\r\n\r\nD:\\myStudio2\\Kafka\\confluent-kafka-dotnet\\examples\\AdvancedConsumer>dotnet run PollWithManualCommit localhost:9092 Advanced\r\nSubscribed to: [Advanced]\r\nCtrl-C to exit.\r\nAssigned partitions: [Advanced [0]], member id: rdkafka-e4310a50-4621-4c6b-9a26-73fc984b7072\r\nTopic: Advanced Partition: 0 Offset: 0 hello\r\nReached end of topic Advanced partition 0, next message will be at offset 1\r\n```\r\n\r\n\r\nBelow is the result of running Producer, which shows the offset 0\r\n\r\n```\r\nD:\\myStudio2\\Kafka\\confluent-kafka-dotnet\\examples\\AdvancedProducer>dotnet run localhost:9092 Advanced\r\n\r\n-----------------------------------------------------------------------\r\nProducer rdkafka#producer-1 producing on topic Advanced.\r\n-----------------------------------------------------------------------\r\nTo create a kafka message with UTF-8 encoded key/value message:\r\n> key value<Enter>\r\nTo create a kafka message with empty key and UTF-8 encoded value:\r\n> value<enter>\r\nCtrl-C to quit.\r\n\r\n> hello\r\nPartition: 0, Offset: 0\r\n>\r\n```\r\n\r\n\r\nI try to avoid Consume() as it might be deprecated based on this API doc: https://docs.confluent.io/current/clients/confluent-kafka-dotnet/api/Confluent.Kafka.Consumer.html#Confluent_Kafka_Consumer_Consume_Confluent_Kafka_Message__System_Int32_\r\n\r\nHow to reproduce\r\n================\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 - [x] Apache Kafka version: 2.12-1.0.1\r\n - [ ] Client configuration:\r\n - [x] Operating system:Windows 10\r\n - [ ] Provide logs (with \"debug\" : \"...\" as necessary in configuration)\r\n - [ ] Provide broker log excerpts\r\n - [ ] Critical issue\r\n"},"comment":{"url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/comments/375109206","html_url":"https://github.com/confluentinc/confluent-kafka-dotnet/issues/470#issuecomment-375109206","issue_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/470","id":375109206,"user":{"login":"mhowlett","id":19174,"avatar_url":"https://avatars2.githubusercontent.com/u/19174?v=4","gravatar_id":"","url":"https://api.github.com/users/mhowlett","html_url":"https://github.com/mhowlett","followers_url":"https://api.github.com/users/mhowlett/followers","following_url":"https://api.github.com/users/mhowlett/following{/other_user}","gists_url":"https://api.github.com/users/mhowlett/gists{/gist_id}","starred_url":"https://api.github.com/users/mhowlett/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mhowlett/subscriptions","organizations_url":"https://api.github.com/users/mhowlett/orgs","repos_url":"https://api.github.com/users/mhowlett/repos","events_url":"https://api.github.com/users/mhowlett/events{/privacy}","received_events_url":"https://api.github.com/users/mhowlett/received_events","type":"User","site_admin":false},"created_at":"2018-03-21T21:56:38Z","updated_at":"2018-03-21T21:56:38Z","author_association":"COLLABORATOR","body":"unable to reproduce, are you sure you're not accidentally auto committing? \r\ni've manually checked a number of variations of committing, not committing."}}
{"actor":{"display_login":"mhowlett"}}