title: How to serialize/deserialize from ksql avro format to c# using confluent platform
body: <p>I am using <code>KsqlDb</code> a table with the following form:<br></p>
  <p><strong>KSQL-DB Query</strong><br>
  <code>create table currency (id integer,name varchar) with (kafka_topic='currency',partitions=1,value_format='avro');</code></p>
  <p><strong>C# model</strong></p>
  <pre><code>public class Currency
  {
      public int Id{get;set;}
      public string Name{get;set;}
  }
  </code></pre>
  <p>Now i want to know how should i write/read data from this topic in C# using the Confluent library:</p>
  <p><strong>Writing</strong></p>
  <pre><code> IProducer&lt;int, Currency&gt; producer=....
  
   Currency cur=new Currency();
  
   Message&lt;int,Currency&gt; message = new Message&lt;int, Currency&gt;
              {
                  Key = msg.Id,
                  Timestamp = new Timestamp(DateTime.UtcNow, TimestampType.CreateTime),
                  Value = msg
              };
   DeliveryResult&lt;int,Currency&gt; delivery =  await this.producer.ProduceAsync(topic,message);
  </code></pre>
  <p><strong>Reading</strong></p>
  <pre><code>IConsumer&lt;int,Currency&gt; iconsumer = new ConsumerBuilder&lt;int, Currency&gt;(config)
                  .SetKeyDeserializer(Deserializers.Int32) //i assume i need to use the id from my dto
                  .SetValueDeserializer(...) //what deserializer
                  .Build();
  
  ConsumeResult&lt;int,Currency&gt; result = consumer.Consume();
  
  Currency message =  // what deserializer JsonSerializer.Deserialize&lt;Currency&gt;(result.Message.Value);
  </code></pre>
  <p>I am not sure how to go about this so i tried looking for serializer. I found this library <a href="https://github.com/AdrianStrugala/AvroConvert/blob/master/tests/KafkaApp/Program.cs" rel="nofollow noreferrer">AvroSerializer</a> , but i do not get where the author fetches the <code>schema</code>.</p>
  <p>Any help on how to read/write to a specific topic that would match with my <code>ksqldb</code> models ?</p>
  <p><strong>Update</strong></p>
  <p>After some research and some answers here i have started using the <code>schemaRegistry</code></p>
  <pre><code>var config = new ConsumerConfig
              {
                  GroupId = kafkaConfig.ConsumerGroup,
                  BootstrapServers = kafkaConfig.ServerUrl,
                  AutoOffsetReset = AutoOffsetReset.Earliest
              };
  var schemaRegistryConfig = new SchemaRegistryConfig
              {
                  Url = kafkaConfig.SchemaRegistryUrl
              };
  var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
  
  IConsumer&lt;int,Currency&gt; consumer = new ConsumerBuilder&lt;int, Currency&gt;(config)
              .SetKeyDeserializer(new AvroDeserializer&lt;int&gt;(schemaRegistry).AsSyncOverAsync())
              .SetValueDeserializer(new AvroDeserializer&lt;Currency&gt;(schemaRegistry).AsSyncOverAsync())
              .Build();
  
  ConsumeResult&lt;int, Currency&gt; result = consumer.Consume();
  </code></pre>
  <p>Now i am getting another error:</p>
  <blockquote>
  <p>Expecting data framing of length 5 bytes or more but total data size
  is 4 bytes</p>
  </blockquote>
  <p>As someone kindly pointed out it seems i retrieving only the id from the schema registry.</p>
  <p>How can i just :  <code>insert into currency (id,name) values (1,3)</code> and retrieve it in C# as a POCO (listed above) ?</p>
  <p><strong>Update 2</strong><Br></p>
  <p>After i have found this <a href="https://github.com/confluentinc/confluent-kafka-dotnet/blob/1.7.x/examples/AvroSpecific/Program.cs" rel="nofollow noreferrer">source</a> program it seems i am not able to publish messages to tables for some reason.</p>
  <p>There is no error when sending the message but it is not published to Kafka.</p>