{"action":"created","comment":{"url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/comments/211412554","pull_request_review_id":147828866,"id":211412554,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDIxMTQxMjU1NA==","diff_hunk":"@@ -0,0 +1,750 @@\n+// Copyright 2018 Confluent Inc.\n+//\n+// Licensed under the Apache License, Version 2.0 (the \"License\");\n+// you may not use this file except in compliance with the License.\n+// You may obtain a copy of the License at\n+//\n+// http://www.apache.org/licenses/LICENSE-2.0\n+//\n+// Unless required by applicable law or agreed to in writing, software\n+// distributed under the License is distributed on an \"AS IS\" BASIS,\n+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+// See the License for the specific language governing permissions and\n+// limitations under the License.\n+//\n+// Refer to LICENSE for more information.\n+\n+using System;\n+using System.Collections.Generic;\n+using System.Collections.Concurrent;\n+using System.Linq;\n+using System.Runtime.InteropServices;\n+using System.Threading;\n+using System.Threading.Tasks;\n+using Confluent.Kafka.Admin;\n+using Confluent.Kafka.Impl;\n+using static Confluent.Kafka.Internal.Util.Marshal;\n+\n+\n+namespace Confluent.Kafka\n+{\n+    /// <summary>\n+    ///     Implements an Apache Kafka admin client.\n+    /// </summary>\n+    public class AdminClient : IAdminClient\n+    {\n+        private Task callbackTask;\n+        private CancellationTokenSource callbackCts;\n+\n+        private IntPtr resultQueue = IntPtr.Zero;\n+\n+        private List<CreateTopicResult> extractTopicResults(IntPtr topicResultsPtr, int topicResultsCount)\n+        {\n+            IntPtr[] topicResultsPtrArr = new IntPtr[topicResultsCount];\n+            Marshal.Copy(topicResultsPtr, topicResultsPtrArr, 0, topicResultsCount);\n+\n+            return topicResultsPtrArr.Select(topicResultPtr => new CreateTopicResult \n+                {\n+                    Topic = PtrToStringUTF8(Librdkafka.topic_result_name(topicResultPtr)),\n+                    Error = new Error(\n+                        Librdkafka.topic_result_error(topicResultPtr), \n+                        PtrToStringUTF8(Librdkafka.topic_result_error_string(topicResultPtr)))\n+                }).ToList();\n+        }\n+\n+        private ConfigEntryResult extractConfigEntry(IntPtr configEntryPtr)\n+        {\n+            var synonyms = new List<ConfigSynonym>();\n+            var synonymsPtr = Librdkafka.ConfigEntry_synonyms(configEntryPtr, out UIntPtr synonymsCount);\n+            if (synonymsPtr != IntPtr.Zero)\n+            {\n+                IntPtr[] synonymsPtrArr = new IntPtr[(int)synonymsCount];\n+                Marshal.Copy(synonymsPtr, synonymsPtrArr, 0, (int)synonymsCount);\n+                synonyms = synonymsPtrArr\n+                    .Select(synonymPtr => extractConfigEntry(synonymPtr))\n+                    .Select(e => new ConfigSynonym { Name = e.Name, Value = e.Value, Source = e.Source } )\n+                    .ToList();\n+            }\n+\n+            return new ConfigEntryResult\n+            {\n+                Name = PtrToStringUTF8(Librdkafka.ConfigEntry_name(configEntryPtr)),\n+                Value = PtrToStringUTF8(Librdkafka.ConfigEntry_value(configEntryPtr)),\n+                IsDefault = (int)Librdkafka.ConfigEntry_is_default(configEntryPtr) == 1,\n+                IsSensitive = (int)Librdkafka.ConfigEntry_is_sensitive(configEntryPtr) == 1,\n+                IsReadOnly = (int)Librdkafka.ConfigEntry_is_read_only(configEntryPtr) == 1,\n+                Source = Librdkafka.ConfigEntry_source(configEntryPtr),\n+                Synonyms = synonyms\n+            };\n+        }\n+\n+        private List<DescribeConfigResult> extractResultConfigs(IntPtr configResourcesPtr, int configResourceCount)\n+        {\n+            var result = new List<DescribeConfigResult>();\n+\n+            IntPtr[] configResourcesPtrArr = new IntPtr[configResourceCount];\n+            Marshal.Copy(configResourcesPtr, configResourcesPtrArr, 0, configResourceCount);\n+            foreach (var configResourcePtr in configResourcesPtrArr)\n+            {\n+                var resourceName = PtrToStringUTF8(Librdkafka.ConfigResource_name(configResourcePtr));\n+                var errorCode = Librdkafka.ConfigResource_error(configResourcePtr);\n+                var errorReason = PtrToStringUTF8(Librdkafka.ConfigResource_error_string(configResourcePtr));\n+                var resourceConfigType = Librdkafka.ConfigResource_type(configResourcePtr);\n+\n+                var configEntriesPtr = Librdkafka.ConfigResource_configs(configResourcePtr, out UIntPtr configEntryCount);\n+                IntPtr[] configEntriesPtrArr = new IntPtr[(int)configEntryCount];\n+                if ((int)configEntryCount > 0)\n+                {\n+                    Marshal.Copy(configEntriesPtr, configEntriesPtrArr, 0, (int)configEntryCount);\n+                }\n+                var configEntries = configEntriesPtrArr\n+                    .Select(configEntryPtr => extractConfigEntry(configEntryPtr))\n+                    .ToDictionary(e => e.Name);\n+\n+                result.Add(new DescribeConfigResult { \n+                    ConfigResource = new ConfigResource { Name = resourceName, Type = resourceConfigType },\n+                    Entries = configEntries,\n+                    Error = new Error(errorCode, errorReason)\n+                });\n+            }\n+\n+            return result;\n+        }\n+\n+        private const int POLL_TIMEOUT_MS = 100;\n+        private Task StartPollTask(CancellationToken ct)\n+            => Task.Factory.StartNew(() =>\n+                {\n+                    try\n+                    {\n+                        while (true)\n+                        {\n+                            ct.ThrowIfCancellationRequested();\n+\n+                            try\n+                            {\n+                                var eventPtr = kafkaHandle.QueuePoll(resultQueue, POLL_TIMEOUT_MS);\n+                                if (eventPtr == IntPtr.Zero)\n+                                {\n+                                    continue;\n+                                }\n+\n+                                var type = Librdkafka.event_type(eventPtr);\n+\n+                                var ptr = (IntPtr)Librdkafka.event_opaque(eventPtr);\n+                                var gch = GCHandle.FromIntPtr(ptr);\n+                                var adminClientResult = gch.Target;\n+                                gch.Free();\n+\n+                                if (!adminClientResultTypes.TryGetValue(type, out Type expectedType))\n+                                {\n+                                    // Should never happen.\n+                                    throw new InvalidOperationException($\"Unknown result type: {type}\");\n+                                }\n+\n+                                if (adminClientResult.GetType() != expectedType)\n+                                {\n+                                    // Should never happen.\n+                                    throw new InvalidOperationException($\"Completion source type mismatch. Exected {expectedType.Name}, got {type}\");\n+                                }\n+\n+                                var errorCode = Librdkafka.event_error(eventPtr);\n+                                var errorStr = Librdkafka.event_error_string(eventPtr);\n+\n+                                switch (type)\n+                                {\n+                                    case Librdkafka.EventType.CreateTopics_Result:\n+                                        {\n+                                            if (errorCode != ErrorCode.NoError)\n+                                            {\n+                                                ((TaskCompletionSource<List<CreateTopicResult>>)adminClientResult).TrySetException(new KafkaException(new Error(errorCode, errorStr)));\n+                                                return;\n+                                            }\n+\n+                                            var result = extractTopicResults(\n+                                                Librdkafka.CreateTopics_result_topics(eventPtr, out UIntPtr resultCountPtr), (int)resultCountPtr);\n+\n+                                            if (result.Any(r => r.Error.IsError))\n+                                            {\n+                                                ((TaskCompletionSource<List<CreateTopicResult>>)adminClientResult).TrySetException(new CreateTopicsException(result));\n+                                            }\n+                                            else\n+                                            {\n+                                                ((TaskCompletionSource<List<CreateTopicResult>>)adminClientResult).TrySetResult(result);\n+                                            }\n+                                        }\n+                                        break;\n+\n+                                    case Librdkafka.EventType.DeleteTopics_Result:\n+                                        {\n+                                            if (errorCode != ErrorCode.NoError)\n+                                            {\n+                                                ((TaskCompletionSource<List<DeleteTopicResult>>)adminClientResult).TrySetException(new KafkaException(new Error(errorCode, errorStr)));\n+                                                return;\n+                                            }\n+\n+                                            var result = extractTopicResults(\n+                                                Librdkafka.DeleteTopics_result_topics(eventPtr, out UIntPtr resultCountPtr), (int)resultCountPtr)\n+                                                    .Select(r => new DeleteTopicResult { Topic = r.Topic, Error = r.Error }).ToList();\n+\n+                                            if (result.Any(r => r.Error.IsError))\n+                                            {\n+                                                ((TaskCompletionSource<List<DeleteTopicResult>>)adminClientResult).TrySetException(new DeleteTopicsException(result));\n+                                            }\n+                                            else\n+                                            {\n+                                                ((TaskCompletionSource<List<DeleteTopicResult>>)adminClientResult).TrySetResult(result);\n+                                            }\n+                                        }\n+                                        break;\n+\n+                                    case Librdkafka.EventType.CreatePartitions_Result:\n+                                        {\n+                                            if (errorCode != ErrorCode.NoError)\n+                                            {\n+                                                ((TaskCompletionSource<List<CreatePartitionsResult>>)adminClientResult).TrySetException(new KafkaException(new Error(errorCode, errorStr)));\n+                                                return;\n+                                            }\n+\n+                                            var result = extractTopicResults(\n+                                                    Librdkafka.CreatePartitions_result_topics(eventPtr, out UIntPtr resultCountPtr), (int)resultCountPtr)\n+                                                        .Select(r => new CreatePartitionsResult { Topic = r.Topic, Error = r.Error }).ToList();\n+\n+                                            if (result.Any(r => r.Error.IsError))\n+                                            {\n+                                                ((TaskCompletionSource<List<CreatePartitionsResult>>)adminClientResult).TrySetException(new CreatePartitionsException(result));\n+                                            }\n+                                            else\n+                                            {\n+                                                ((TaskCompletionSource<List<CreatePartitionsResult>>)adminClientResult).TrySetResult(result);\n+                                            }\n+                                        }\n+                                        break;\n+\n+                                    case Librdkafka.EventType.DescribeConfigs_Result:\n+                                        {\n+                                            if (errorCode != ErrorCode.NoError)\n+                                            {\n+                                                ((TaskCompletionSource<List<DescribeConfigResult>>)adminClientResult).TrySetException(new KafkaException(new Error(errorCode, errorStr)));\n+                                                return;\n+                                            }\n+\n+                                            var result = extractResultConfigs(\n+                                                Librdkafka.DescribeConfigs_result_resources(eventPtr, out UIntPtr cntp), (int)cntp);\n+\n+                                            if (result.Any(r => r.Error.IsError))\n+                                            {\n+                                                ((TaskCompletionSource<List<DescribeConfigResult>>)adminClientResult).TrySetException(new DescribeConfigsException(result));\n+                                            }\n+                                            else\n+                                            {\n+                                                ((TaskCompletionSource<List<DescribeConfigResult>>)adminClientResult).TrySetResult(result);\n+                                            }\n+                                        }\n+                                        break;\n+\n+                                    case Librdkafka.EventType.AlterConfigs_Result:\n+                                        {\n+                                            if (errorCode != ErrorCode.NoError)\n+                                            {\n+                                                ((TaskCompletionSource<List<AlterConfigResult>>)adminClientResult).TrySetException(new KafkaException(new Error(errorCode, errorStr)));\n+                                                return;\n+                                            }\n+\n+                                            var result = extractResultConfigs(\n+                                                Librdkafka.AlterConfigs_result_resources(eventPtr, out UIntPtr cntp), (int)cntp)\n+                                                    .Select(r => new AlterConfigResult { ConfigResource = r.ConfigResource, Error = r.Error }).ToList();\n+\n+                                            if (result.Any(r => r.Error.IsError))\n+                                            {\n+                                                ((TaskCompletionSource<List<AlterConfigResult>>)adminClientResult).TrySetException(new AlterConfigsException(result));\n+                                            }\n+                                            else\n+                                            {\n+                                                ((TaskCompletionSource<List<AlterConfigResult>>) adminClientResult).TrySetResult(result);\n+                                            }\n+                                        }\n+                                        break;\n+\n+                                    default:\n+                                        // Should never happen.\n+                                        throw new InvalidOperationException($\"Unknown result type: {type}\");\n+                                }\n+                            }\n+                            catch\n+                            {\n+                                // TODO: If this occurs, it means there's an application logic error\n+                                //       (i.e. program execution should never get here). Rather than\n+                                //       ignore the situation, we panic, destroy the librdkafka handle, \n+                                //       and exit the polling loop. Further usage of the AdminClient will\n+                                //       result in exceptions. People will be sure to notice and tell us.\n+                                this.DisposeResources();\n+                            }\n+                        }\n+                    }\n+                    catch (OperationCanceledException) {}\n+                }, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default);\n+\n+\n+        internal static Dictionary<Librdkafka.EventType, Type> adminClientResultTypes = new Dictionary<Librdkafka.EventType, Type>\n+        {\n+            { Librdkafka.EventType.CreateTopics_Result, typeof(TaskCompletionSource<List<CreateTopicResult>>) },\n+            { Librdkafka.EventType.DeleteTopics_Result, typeof(TaskCompletionSource<List<DeleteTopicResult>>) },\n+            { Librdkafka.EventType.DescribeConfigs_Result, typeof(TaskCompletionSource<List<DescribeConfigResult>>) },\n+            { Librdkafka.EventType.AlterConfigs_Result, typeof(TaskCompletionSource<List<AlterConfigResult>>) },\n+            { Librdkafka.EventType.CreatePartitions_Result, typeof(TaskCompletionSource<List<CreatePartitionsResult>>) }\n+        };\n+\n+        /// <summary>\n+        ///     Get the configuration for the specified resources. The returned \n+        ///     configuration includes default values and the IsDefault property\n+        ///     can be used to distinguish them from user supplied values. The \n+        ///     value of config entries where IsSensitive is true is always null \n+        ///     so that sensitive information is not disclosed. Config entries where\n+        ///     IsReadOnly is true cannot be updated. This operation is supported \n+        ///     by brokers with version 0.11.0.0 or higher.\n+        /// </summary>\n+        /// <param name=\"resources\">\n+        ///     The resources (topic and broker resource types are currently \n+        ///     supported)\n+        /// </param>\n+        /// <param name=\"options\">\n+        ///     The options to use when describing configs.\n+        /// </param>\n+        /// <returns>\n+        ///     Configs for the specified resources.\n+        /// </returns>\n+        public Task<List<DescribeConfigResult>> DescribeConfigsAsync(IEnumerable<ConfigResource> resources, DescribeConfigsOptions options = null)\n+        {\n+            // TODO: To support results that may complete at different times, we may also want to implement:\n+            // List<Task<DescribeConfigResult>> DescribeConfigsConcurrent(IEnumerable<ConfigResource> resources, DescribeConfigsOptions options = null)\n+\n+            var completionSource = new TaskCompletionSource<List<DescribeConfigResult>>();\n+            var gch = GCHandle.Alloc(completionSource);\n+            Handle.LibrdkafkaHandle.DescribeConfigs(\n+                resources, options, resultQueue,\n+                GCHandle.ToIntPtr(gch));\n+            return completionSource.Task;\n+        }\n+\n+        /// <summary>\n+        ///     Update the configuration for the specified resources. Updates are not transactional\n+        ///     so they may succeed for some resources while fail for others. The configs for a \n+        ///     particular resource are updated atomically. This operation is supported by brokers \n+        ///     with version 0.11.0.0 or higher.\n+        /// </summary>\n+        /// <param name=\"configs\">\n+        ///     The resources with their configs (topic is the only resource type with configs\n+        ///     that can be updated currently)\n+        /// </param>\n+        /// <param name=\"options\">\n+        ///     The options to use when altering configs.\n+        /// </param>\n+        /// <returns>\n+        ///     The results of the alter configs requests.\n+        /// </returns>\n+        public Task<List<AlterConfigResult>> AlterConfigsAsync(Dictionary<ConfigResource, List<ConfigEntry>> configs, AlterConfigsOptions options = null)\n+        {\n+            // TODO: To support results that may complete at different times, we may also want to implement:\n+            // List<Task<AlterConfigResult>> AlterConfigsConcurrent(Dictionary<ConfigResource, Config> configs, AlterConfigsOptions options = null)\n+\n+            var completionSource = new TaskCompletionSource<List<AlterConfigResult>>();\n+            // Note: There is a level of indirection between the GCHandle and\n+            // physical memory address. GCHangle.ToIntPtr doesn't return the\n+            // physical address, it returns an id that refers to the object via\n+            // a handle-table.\n+            var gch = GCHandle.Alloc(completionSource);\n+            Handle.LibrdkafkaHandle.AlterConfigs(\n+                configs, options, resultQueue,\n+                GCHandle.ToIntPtr(gch));\n+            return completionSource.Task;\n+        }\n+\n+        /// <summary>\n+        ///     Create a batch of new topics.\n+        /// </summary>\n+        /// <param name=\"topics\">\n+        ///     A collection of specifications of the new topics to create.\n+        /// </param>\n+        /// <param name=\"options\">\n+        ///     The options to use when creating the topics.\n+        /// </param>\n+        /// <returns>\n+        ///     The results of the create topic requests.\n+        /// </returns>\n+        public Task<List<CreateTopicResult>> CreateTopicsAsync(IEnumerable<NewTopic> topics, CreateTopicsOptions options = null)\n+        {\n+            // TODO: To support results that may complete at different times, we may also want to implement:\n+            // public List<Task<CreateTopicResult>> CreateTopicsConcurrent(IEnumerable<NewTopic> topics, CreateTopicsOptions options = null)\n+\n+            var completionSource = new TaskCompletionSource<List<CreateTopicResult>>();\n+            var gch = GCHandle.Alloc(completionSource);\n+            Handle.LibrdkafkaHandle.CreateTopics(\n+                topics, options, resultQueue,\n+                GCHandle.ToIntPtr(gch));\n+            return completionSource.Task;\n+        }\n+\n+        /// <summary>\n+        ///     Delete a batch of topics. This operation is not transactional so it may succeed for some topics while fail\n+        ///     for others. It may take several seconds after the DeleteTopicsResult returns success for all the brokers to\n+        ///     become aware that the topics are gone. During this time, topics may continue to be visible via admin \n+        ///     operations. If delete.topic.enable is false on the brokers, DeleteTopicsAsync will mark the topics for\n+        ///     deletion, but not actually delete them. The futures will return successfully in this case.\n+        /// </summary>\n+        /// <param name=\"topics\">\n+        ///     The topic names to delete.\n+        /// </param>\n+        /// <param name=\"options\">\n+        ///     The options to use when deleting topics.\n+        /// </param>\n+        /// <returns>\n+        ///     The results of the delete topic requests.\n+        /// </returns>\n+        public Task<List<DeleteTopicResult>> DeleteTopicsAsync(IEnumerable<string> topics, DeleteTopicsOptions options = null)\n+        {\n+            // TODO: To support results that may complete at different times, we may also want to implement:\n+            // List<Task<DeleteTopicResult>> DeleteTopicsConcurrent(IEnumerable<string> topics, DeleteTopicsOptions options = null)\n+\n+            var completionSource = new TaskCompletionSource<List<DeleteTopicResult>>();\n+            var gch = GCHandle.Alloc(completionSource);\n+            Handle.LibrdkafkaHandle.DeleteTopics(\n+                topics, options, resultQueue,\n+                GCHandle.ToIntPtr(gch));\n+            return completionSource.Task;\n+        }\n+\n+        /// <summary>\n+        ///     Increase the number of partitions for one or more topics as per\n+        ///     the supplied NewPartitions specifications.\n+        /// </summary>\n+        /// <param name=\"newPartitions\">\n+        ///     A collection of NewPartitions specifications.\n+        /// </param>\n+        /// <param name=\"options\">\n+        ///     The options to use when creating the partitions.\n+        /// </param>\n+        /// <returns>\n+        ///     The results of the NewPartitions requests.\n+        /// </returns>\n+        public Task<List<CreatePartitionsResult>> CreatePartitionsAsync(\n+            IEnumerable<NewPartitions> newPartitions, CreatePartitionsOptions options = null)\n+        {\n+            // TODO: To support results that may complete at different times, we may also want to implement:\n+            // List<Task<CreatePartitionResult>> CreatePartitionsConcurrent(IEnumerable<NewPartitions> newPartitions, CreatePartitionsOptions options = null)\n+\n+            var completionSource = new TaskCompletionSource<List<CreatePartitionsResult>>();\n+            var gch = GCHandle.Alloc(completionSource);\n+            Handle.LibrdkafkaHandle.CreatePartitions(\n+                newPartitions, options, resultQueue,\n+                GCHandle.ToIntPtr(gch));\n+            return completionSource.Task;\n+        }\n+\n+        private IClient ownedClient;\n+        private Handle handle;\n+\n+        private SafeKafkaHandle kafkaHandle\n+            => handle.LibrdkafkaHandle;\n+\n+        private ConcurrentDictionary<string, SafeTopicHandle> topicHandles\n+            = new ConcurrentDictionary<string, SafeTopicHandle>(StringComparer.Ordinal);\n+\n+        private Func<string, SafeTopicHandle> topicHandlerFactory;\n+\n+\n+        /// <summary>\n+        ///     Initialize a new AdminClient instance.\n+        /// </summary>\n+        /// <param name=\"config\">\n+        ///     A collection of librdkafka configuration parameters \n+        ///     (refer to https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md)\n+        ///     and parameters specific to this client (refer to: \n+        ///     <see cref=\"Confluent.Kafka.ConfigPropertyNames\" />)\n+        /// </param>\n+        public AdminClient(IEnumerable<KeyValuePair<string, object>> config)\n+        {\n+            // TODO: Sort out exact behavior with poll. We don't really want to\n+            // have the capability to manually poll on this class because we\n+            // don't know if we're wrapping a consumer or producer.\n+\n+            this.ownedClient = new Producer(config);\n+            this.handle = new Handle\n+            { \n+                Owner = this,\n+                LibrdkafkaHandle = ownedClient.Handle.LibrdkafkaHandle\n+            };\n+            Init();\n+        }\n+\n+\n+        /// <summary>\n+        ///     Initialize a new AdminClient instance.\n+        /// </summary>\n+        /// <param name=\"handle\">\n+        ///     An underlying librdkafka client handle to use to make broker requests.\n+        ///     It is valid to provide either a Consumer or Producer handle.\n+        /// </param>\n+        public AdminClient(Handle handle)\n+        {\n+            this.ownedClient = null;\n+            this.handle = handle;\n+            Init();\n+        }\n+\n+        private void Init()\n+        {\n+            // note: ConcurrentDictionary.GetorAdd() method is not atomic\n+            this.topicHandlerFactory = (string topicName) =>\n+            {\n+                // Note: there is a possible (benign) race condition here - topicHandle could have already\n+                // been created for the topic (and possibly added to topicHandles). If the topicHandle has\n+                // already been created, rdkafka will return it and not create another. the call to rdkafka","path":"src/Confluent.Kafka/AdminClient.cs","position":502,"original_position":502,"commit_id":"3ded651a2b03556a14b8f47b20b2db2d249b7410","original_commit_id":"a825c18ba71483990e90d5d8d3906c53f1caa60a","user":{"login":"mhowlett","id":19174,"node_id":"MDQ6VXNlcjE5MTc0","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},"body":"good point. replaced this with code that does custom locking. note: which is now in SafeKafkaHandle.cs. i don't know why this is implemented this way.","created_at":"2018-08-20T21:25:52Z","updated_at":"2018-08-20T21:25:52Z","html_url":"https://github.com/confluentinc/confluent-kafka-dotnet/pull/582#discussion_r211412554","pull_request_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/582","author_association":"COLLABORATOR","_links":{"self":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/comments/211412554"},"html":{"href":"https://github.com/confluentinc/confluent-kafka-dotnet/pull/582#discussion_r211412554"},"pull_request":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/582"}},"in_reply_to_id":208582801},"pull_request":{"url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/582","id":206502577,"node_id":"MDExOlB1bGxSZXF1ZXN0MjA2NTAyNTc3","html_url":"https://github.com/confluentinc/confluent-kafka-dotnet/pull/582","diff_url":"https://github.com/confluentinc/confluent-kafka-dotnet/pull/582.diff","patch_url":"https://github.com/confluentinc/confluent-kafka-dotnet/pull/582.patch","issue_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/582","number":582,"state":"open","locked":false,"title":"1.0 API","user":{"login":"mhowlett","id":19174,"node_id":"MDQ6VXNlcjE5MTc0","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},"body":"@edenhill - it's about time we got this on master! have at it.","created_at":"2018-08-06T20:05:32Z","updated_at":"2018-08-20T21:25:52Z","closed_at":null,"merged_at":null,"merge_commit_sha":"efd02d5fe5b27844ad3dd40aaa68242e5f1b85c0","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/582/commits","review_comments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/582/comments","review_comment_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/comments{/number}","comments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/582/comments","statuses_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/statuses/3ded651a2b03556a14b8f47b20b2db2d249b7410","head":{"label":"confluentinc:1.0-experimental","ref":"1.0-experimental","sha":"3ded651a2b03556a14b8f47b20b2db2d249b7410","user":{"login":"confluentinc","id":9439498,"node_id":"MDEyOk9yZ2FuaXphdGlvbjk0Mzk0OTg=","avatar_url":"https://avatars0.githubusercontent.com/u/9439498?v=4","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"repo":{"id":72554578,"node_id":"MDEwOlJlcG9zaXRvcnk3MjU1NDU3OA==","name":"confluent-kafka-dotnet","full_name":"confluentinc/confluent-kafka-dotnet","owner":{"login":"confluentinc","id":9439498,"node_id":"MDEyOk9yZ2FuaXphdGlvbjk0Mzk0OTg=","avatar_url":"https://avatars0.githubusercontent.com/u/9439498?v=4","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/confluentinc/confluent-kafka-dotnet","description":"Confluent's Apache Kafka .NET client","fork":false,"url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet","forks_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/forks","keys_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/keys{/key_id}","collaborators_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/teams","hooks_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/hooks","issue_events_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/events{/number}","events_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/events","assignees_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/assignees{/user}","branches_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/branches{/branch}","tags_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/tags","blobs_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/refs{/sha}","trees_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/trees{/sha}","statuses_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/statuses/{sha}","languages_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/languages","stargazers_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/stargazers","contributors_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/contributors","subscribers_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/subscribers","subscription_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/subscription","commits_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/commits{/sha}","git_commits_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/commits{/sha}","comments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/comments{/number}","issue_comment_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/comments{/number}","contents_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/contents/{+path}","compare_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/compare/{base}...{head}","merges_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/merges","archive_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/downloads","issues_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues{/number}","pulls_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls{/number}","milestones_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/milestones{/number}","notifications_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/labels{/name}","releases_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/releases{/id}","deployments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/deployments","created_at":"2016-11-01T16:25:52Z","updated_at":"2018-08-20T10:20:38Z","pushed_at":"2018-08-20T17:04:02Z","git_url":"git://github.com/confluentinc/confluent-kafka-dotnet.git","ssh_url":"git@github.com:confluentinc/confluent-kafka-dotnet.git","clone_url":"https://github.com/confluentinc/confluent-kafka-dotnet.git","svn_url":"https://github.com/confluentinc/confluent-kafka-dotnet","homepage":"https://docs.confluent.io/current/clients/confluent-kafka-dotnet/api/Confluent.Kafka.html","size":1535,"stargazers_count":615,"watchers_count":615,"language":"C#","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":188,"mirror_url":null,"archived":false,"open_issues_count":56,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":188,"open_issues":56,"watchers":615,"default_branch":"master"}},"base":{"label":"confluentinc:master","ref":"master","sha":"0a66d2697704d0351211d1fed63ebeb73d5ab2b1","user":{"login":"confluentinc","id":9439498,"node_id":"MDEyOk9yZ2FuaXphdGlvbjk0Mzk0OTg=","avatar_url":"https://avatars0.githubusercontent.com/u/9439498?v=4","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"repo":{"id":72554578,"node_id":"MDEwOlJlcG9zaXRvcnk3MjU1NDU3OA==","name":"confluent-kafka-dotnet","full_name":"confluentinc/confluent-kafka-dotnet","owner":{"login":"confluentinc","id":9439498,"node_id":"MDEyOk9yZ2FuaXphdGlvbjk0Mzk0OTg=","avatar_url":"https://avatars0.githubusercontent.com/u/9439498?v=4","gravatar_id":"","url":"https://api.github.com/users/confluentinc","html_url":"https://github.com/confluentinc","followers_url":"https://api.github.com/users/confluentinc/followers","following_url":"https://api.github.com/users/confluentinc/following{/other_user}","gists_url":"https://api.github.com/users/confluentinc/gists{/gist_id}","starred_url":"https://api.github.com/users/confluentinc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/confluentinc/subscriptions","organizations_url":"https://api.github.com/users/confluentinc/orgs","repos_url":"https://api.github.com/users/confluentinc/repos","events_url":"https://api.github.com/users/confluentinc/events{/privacy}","received_events_url":"https://api.github.com/users/confluentinc/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/confluentinc/confluent-kafka-dotnet","description":"Confluent's Apache Kafka .NET client","fork":false,"url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet","forks_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/forks","keys_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/keys{/key_id}","collaborators_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/teams","hooks_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/hooks","issue_events_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/events{/number}","events_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/events","assignees_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/assignees{/user}","branches_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/branches{/branch}","tags_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/tags","blobs_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/refs{/sha}","trees_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/trees{/sha}","statuses_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/statuses/{sha}","languages_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/languages","stargazers_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/stargazers","contributors_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/contributors","subscribers_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/subscribers","subscription_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/subscription","commits_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/commits{/sha}","git_commits_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/git/commits{/sha}","comments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/comments{/number}","issue_comment_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/comments{/number}","contents_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/contents/{+path}","compare_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/compare/{base}...{head}","merges_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/merges","archive_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/downloads","issues_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues{/number}","pulls_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls{/number}","milestones_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/milestones{/number}","notifications_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/labels{/name}","releases_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/releases{/id}","deployments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/deployments","created_at":"2016-11-01T16:25:52Z","updated_at":"2018-08-20T10:20:38Z","pushed_at":"2018-08-20T17:04:02Z","git_url":"git://github.com/confluentinc/confluent-kafka-dotnet.git","ssh_url":"git@github.com:confluentinc/confluent-kafka-dotnet.git","clone_url":"https://github.com/confluentinc/confluent-kafka-dotnet.git","svn_url":"https://github.com/confluentinc/confluent-kafka-dotnet","homepage":"https://docs.confluent.io/current/clients/confluent-kafka-dotnet/api/Confluent.Kafka.html","size":1535,"stargazers_count":615,"watchers_count":615,"language":"C#","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":188,"mirror_url":null,"archived":false,"open_issues_count":56,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":188,"open_issues":56,"watchers":615,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/582"},"html":{"href":"https://github.com/confluentinc/confluent-kafka-dotnet/pull/582"},"issue":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/582"},"comments":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/issues/582/comments"},"review_comments":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/582/comments"},"review_comment":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/pulls/582/commits"},"statuses":{"href":"https://api.github.com/repos/confluentinc/confluent-kafka-dotnet/statuses/3ded651a2b03556a14b8f47b20b2db2d249b7410"}},"author_association":"COLLABORATOR"}}
{"actor":{"display_login":"mhowlett"}}