@@ -37,50 +37,29 @@ static void Main(string[] args)
3737 string schemaRegistryUrl = args [ 1 ] ;
3838 string topicName = args [ 2 ] ;
3939
40- var producerConfig = new Dictionary < string , object >
40+ var producerConfig = new ProducerConfig { BootstrapServers = bootstrapServers } ;
41+
42+ var avroConfig = new AvroSerdeProviderConfig
4143 {
42- { "bootstrap.servers" , bootstrapServers } ,
4344 // Note: you can specify more than one schema registry url using the
4445 // schema.registry.url property for redundancy (comma separated list).
4546 // The property name is not plural to follow the convention set by
4647 // the Java implementation.
47- { "schema.registry.url" , schemaRegistryUrl } ,
48+ SchemaRegistryUrl = schemaRegistryUrl ,
4849 // optional schema registry client properties:
49- { "schema.registry.connection.timeout.ms" , 5000 } ,
50- { "schema.registry.max.cached.schemas" , 10 } ,
50+ SchemaRegistryRequestTimeoutMs = 5000 ,
51+ SchemaRegistryMaxCachedSchemas = 10 ,
5152 // optional avro serializer properties:
52- { "avro.serializer.buffer.bytes" , 50 } ,
53- { "avro.serializer.auto.register.schemas" , true }
53+ AvroSerializerBufferBytes = 50 ,
54+ AvroSerializerAutoRegisterSchemas = true
5455 } ;
5556
56- var consumerConfig = new Dictionary < string , object >
57+ var consumerConfig = new ConsumerConfig
5758 {
58- { "bootstrap.servers" , bootstrapServers } ,
59- { "group.id" , Guid . NewGuid ( ) } ,
60- { "schema.registry.url" , schemaRegistryUrl } ,
61- { "error_cb" , ( Action < ErrorEvent > ) ( e => Console . WriteLine ( $ "Error [{ e . Level } ]: { e . Error . Reason } ") ) }
59+ BootstrapServers = bootstrapServers ,
60+ GroupId = Guid . NewGuid ( ) . ToString ( )
6261 } ;
6362
64- // Note: Each AvroSerializer and AvroDeserializer instance created below internally creates and manages
65- // the lifecycle of a CachedSchemaRegistry instance, taking it's configuration properties from the
66- // dictionary supplied to the Producer/Consumer constructor. This is the most straightforward way
67- // to use these (de)serializers, however you can also pass a CachedSchemaRegistry instance into the
68- // constructors directly, which avoids duplicate connections to Schema Registry in the case where
69- // you're using more than one AvroSerializer/Deserializer. E.g.:
70- //
71- // var schemaRegistryConfig = new Dictionary<string, object>
72- // {
73- // { "schema.registry.url", schemaRegistryUrl },
74- // { "schema.registry.connection.timeout.ms", 5000 }, // optional
75- // { "schema.registry.max.cached.schemas", 10 } // optional
76- // };
77- //
78- // using (var schemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryConfig))
79- // using (var consumer = new Consumer<int, User>(consumerConfig, new AvroDeserializer<int>(schemaRegistryClient), new AvroDeserializer<User>(schemaRegistryClient)))
80- // using (var producer = new Producer<int, User>(producerConfig, new AvroSerializer<int>(schemaRegistryClient), new AvroSerializer<User>(schemaRegistryClient)))
81- // {
82- // ...
83-
8463 // Note: The User class in this project was generated using the Confluent fork of the avrogen.exe tool
8564 // (avaliable from: https://github.com/confluentinc/avro/tree/confluent-fork) which includes modifications
8665 // that prevent namespace clashes with user namespaces that include the identifier 'Avro'. AvroSerializer
@@ -90,8 +69,12 @@ static void Main(string[] args)
9069 CancellationTokenSource cts = new CancellationTokenSource ( ) ;
9170 var consumeTask = Task . Run ( ( ) =>
9271 {
93- using ( var consumer = new Consumer < string , User > ( consumerConfig , new AvroDeserializer < string > ( ) , new AvroDeserializer < User > ( ) ) )
72+ using ( var serdeProvider = new AvroSerdeProvider ( avroConfig ) )
73+ using ( var consumer = new Consumer < string , User > ( consumerConfig , serdeProvider . CreateDeserializer < string > ( ) , serdeProvider . CreateDeserializer < User > ( ) ) )
9474 {
75+ consumer . OnError += ( _ , e )
76+ => Console . WriteLine ( $ "Error: { e . Reason } ") ;
77+
9578 consumer . Subscribe ( topicName ) ;
9679
9780 while ( ! cts . Token . IsCancellationRequested )
@@ -111,7 +94,8 @@ static void Main(string[] args)
11194 }
11295 } , cts . Token ) ;
11396
114- using ( var producer = new Producer < string , User > ( producerConfig , new AvroSerializer < string > ( ) , new AvroSerializer < User > ( ) ) )
97+ using ( var serdeProvider = new AvroSerdeProvider ( avroConfig ) )
98+ using ( var producer = new Producer < string , User > ( producerConfig , serdeProvider . CreateKeySerializer < string > ( ) , serdeProvider . CreateValueSerializer < User > ( ) ) )
11599 {
116100 Console . WriteLine ( $ "{ producer . Name } producing on { topicName } . Enter user names, q to exit.") ;
117101
0 commit comments