Facing issue with streaming of a topic with Avro Serializer

[shashi812]

Below is the stack trace:
Method not found: 'Void Confluent.SchemaRegistry.Serdes.AvroSerializer`1..ctor(Confluent.SchemaRegistry.ISchemaRegistryClient, Confluent.SchemaRegistry.Serdes.AvroSerializerConfig)

at Streamiz.Kafka.Net.SchemaRegistry.SerDes.Avro.SchemaAvroSerDes`1.Initialize(SerDesContext context)
at Streamiz.Kafka.Net.Processors.Internal.InternalTopologyBuilder.RewriteTopology(IStreamConfig config)
at Streamiz.Kafka.Net.KafkaStream..ctor(Topology topology, IStreamConfig configuration, IKafkaSupplier kafkaSupplier)
at Streamiz.Kafka.Net.KafkaStream..ctor(Topology topology, IStreamConfig configuration)

Code:

var testKeySpecificAvroSerde = new SchemaAvroSerDes();

var testSpecificAvroSerde = new SchemaAvroSerDes();

var peopleKeySerde = new GuidSerDes();
var peopleSpecificAvroSerde = new SchemaAvroSerDes();

var streamConfig = new StreamConfig<GuidSerDes, SchemaAvroSerDes>();
StreamBuilder _builder = new StreamBuilder();

var peoples = _builder.Stream("people-topic", peopleKeySerde, peopleSpecificAvroSerde);

var tests = _builder.GlobalTable<TestKey, TestTarget>(
"test-dev", InMemory.As<TestKey, TestTarget>("test-table-dev")
.WithKeySerdes(testKeySpecificAvroSerde)
.WithValueSerdes(testSpecificAvroSerde));

TestKey mapper(Guid key, People people) =>
new()
{
company_id = people.company_id,
property_id = people.property_id,
source_id = people.source_id
};

People joiner(People people, TestTarget test)
{
people.organization_topic = test.organization_topic;
return people;
}

// Pluck topic name from message
Func<Guid, People, IRecordContext, string> nameExtractor = (peopleKey, people, recordContext) =>
{
if (string.IsNullOrEmpty(people.organization_topic))
{
throw new InvalidOperationException("Topic name is unresolved");
}
return people.organization_topic;
};

if (_logger.IsEnabled(LogLevel.Debug))
{
guestCards.Peek((k, v) => _logger.LogDebug($"Incoming people {k} with join key {mapper(k, v)}"));
}

// Join
var peopleWithTopic = peoples.Join(tests, mapper, joiner, "test-join-dev");
peopleWithTopic.Peek((k, v) => _logger.LogInformation($"Forwarding people {k} to org topic {v.organization_topic ?? "not found"}"));

// Route to dynamically resolved organization topic
peopleWithTopic.To(nameExtractor, peopleKeySerde, peopleSpecificAvroSerde, "test-producer-dev");

var config = new StreamConfig<GuidSerDes, SchemaAvroSerDes>
{
ApplicationId = "test-app",
BootstrapServers = _configuration["PublicClusterConfig:BootstrapServers"],
SchemaRegistryUrl = _configuration["PublicClusterConfig:SchemaRegistryUrl"],
DefaultKeySerDes = new GuidSerDes(),
DefaultValueSerDes = new SchemaAvroSerDes(),
AutoOffsetReset = AutoOffsetReset.Earliest,
CommitIntervalMs = 5000,
NumStreamThreads = 1,
SecurityProtocol = SecurityProtocol.SaslPlaintext,
SaslMechanism = SaslMechanism.Plain,
SaslUsername = _configuration["PublicClusterConfig:SaslUsername"],
SaslPassword = _configuration["PublicClusterConfig:SaslPassword"],
Logger = LoggerFactory.Create((b) =>
{
b.AddConsole();
b.SetMinimumLevel(LogLevel.Information);
}),
};
var topology = _builder.Build();
var stream = new KafkaStream(topology, config);

Console.CancelKeyPress += (o, e) => stream.Dispose();

await stream.StartAsync(stoppingToken);

[LGouellec]

Hey @shashi812 ,

Which version of Streamiz and Confluent Kafka are you using ?

[shashi812]

Issue occured because of different versions of Streamiz and Confluent Kafka. Fixed it by using correct versions

[LGouellec]

var testKeySpecificAvroSerde = new SchemaAvroSerDes();
var testSpecificAvroSerde = new SchemaAvroSerDes();

Which class are you using ? A custom class ?
Normally you should pass a class type ..

public class SchemaAvroSerDes<T> { ... }