Added Default Client Id required for version 0.9 of Kafka Protocol.

src/Kafka.Basic/KafkaClient.cs

@@ -4,7 +4,7 @@ namespace Kafka.Basic
 {
     public interface IKafkaClient : IDisposable
     {
-        IKafkaTopic Topic(string name);
+        IKafkaTopic Topic(string name, ProducerConfig config = null);
         IKafkaConsumer Consumer(string groupName);
         IKafkaConsumer Consumer(ConsumerOptions options);
         IKafkaSimpleConsumer SimpleConsumer();
@@ -19,9 +19,9 @@ public KafkaClient(string zkConnect)
             _zkConnection = new ZookeeperConnection(zkConnect);
         }
 
-        public IKafkaTopic Topic(string name)
+        public IKafkaTopic Topic(string name, ProducerConfig config = null)
         {
-            return new KafkaTopic(_zkConnection, name);
+            return new KafkaTopic(_zkConnection, name, config);
         }
 
         public IKafkaConsumer Consumer(string groupName)

src/Kafka.Basic/KafkaTopic.cs

@@ -18,14 +18,16 @@ public class KafkaTopic : IKafkaTopic
         private readonly IProducer<string, KafkaMessage> _producer;
         private readonly IZookeeperClient _zkClient;
 
-        public KafkaTopic(IZookeeperConnection zkConnect, string name)
+        public KafkaTopic(IZookeeperConnection zkConnect, string name, ProducerConfig producerConfig = null)
         {
+            if (producerConfig == null)
+                producerConfig = ProducerConfig.GetDefaultConfig();
             _zkConnect = zkConnect;
             _name = name;
             _zkClient = zkConnect.CreateClient();
-            _producer = _zkClient.CreateProducer<string, KafkaMessage>();
+            _producer = _zkClient.CreateProducer<string, KafkaMessage>(producerConfig);
         }
-
+        
         public void Send(params Message[] messages)
         {
             _producer.Send(

src/Kafka.Basic/ProducerConfig.cs

@@ -3,14 +3,20 @@
     public class ProducerConfig
     {
         public const short DefaultAcks = 1;
+        public const string DefaultClientId = "DefaultKafkaBasicClientId";
+        public const int DefaultSendTimeout = 1000;
 
         public short Acks { get; set; }
+        public string ClientId { get; set; }
+        public int SendTimeout { get; set; }
 
-        public static ProducerConfig Default()
+        public static ProducerConfig GetDefaultConfig()
         {
             return new ProducerConfig
             {
-                Acks = DefaultAcks
+                Acks = DefaultAcks,
+                ClientId = DefaultClientId,
+                SendTimeout = DefaultSendTimeout,
             };
         }
 

src/Kafka.Basic/Properties/AssemblyInfo.cs

@@ -32,6 +32,6 @@
 // You can specify all the values or you can default the Build and Revision Numbers 
 // by using the '*' as shown below:
 // [assembly: AssemblyVersion("1.0.*")]
-[assembly: AssemblyVersion("1.0.0.0")]
-[assembly: AssemblyFileVersion("1.0.0.0")]
+[assembly: AssemblyVersion("1.0.0.1")]
+[assembly: AssemblyFileVersion("1.0.0.1")]
 [assembly: InternalsVisibleTo("Kafka.Basic.Test")]

src/Kafka.Basic/ZookeeperClient.cs

@@ -37,7 +37,7 @@ public IEnumerable<Broker> GetAllBrokers()
 
         public IProducer<TKey, TMessage> CreateProducer<TKey, TMessage>()
         {
-            return CreateProducer<TKey, TMessage>(ProducerConfig.Default());
+            return CreateProducer<TKey, TMessage>(ProducerConfig.GetDefaultConfig());
         }
 
         public IProducer<TKey, TMessage> CreateProducer<TKey, TMessage>(ProducerConfig config)
@@ -52,7 +52,9 @@ public IProducer<TKey, TMessage> CreateProducer<TKey, TMessage>(ProducerConfig c
                     }).ToList()
                 )
             {
-                RequiredAcks = config.Acks
+                RequiredAcks = config.Acks,
+                ClientId = config.ClientId,
+                SendTimeout = config.SendTimeout
             };
 
             return new Producer<TKey, TMessage>(producerConfiguration);