search

kmcclellan / kafka-di

An extension of Confluent's Kafka client for use with Microsoft.Extensions.DependencyInjection (and friends).
https://www.nuget.org/packages/Confluent.Kafka.DependencyInjection
MIT License
13 stars 5 forks source link

Error using `AvroSerializer`: Unable to activate type. The following constructors are ambiguous... #2

Open phspies opened 3 years ago

phspies commented 3 years ago

Trying to use this library and it seems I'm missing something. I have the following code:

              services.AddKafkaClient(new Dictionary<string, string>
                {
                    { "bootstrap.servers", Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP") },
                    { "enable.idempotence", Environment.GetEnvironmentVariable("KAFKA_IDEMPOTENCE") },
                    { "group.id", Environment.GetEnvironmentVariable("KAFKA_GROUP") }
                });
                services.AddSingleton<ISchemaRegistryClient>(sp =>
                    new CachedSchemaRegistryClient(new SchemaRegistryConfig
                    {
                        Url = "localhost:8081"
                    }));

Everything works well when I serialize the objects with Newtonjson before the message creation, but it seems the following code breaks the object serialization internal to Kafka.

services.AddSingleton(typeof(IAsyncSerializer<>), typeof(AvroSerializer<>));

This line produces the following exception:

System.InvalidOperationException: Unable to activate type 'Confluent.SchemaRegistry.Serdes.AvroSerializer`1[Confluent.Kafka.Null]'. The following constructors are ambiguous:
Void .ctor(Confluent.SchemaRegistry.ISchemaRegistryClient, System.Collections.Generic.IEnumerable`1[System.Collections.Generic.KeyValuePair`2[System.String,System.String]])
Void .ctor(Confluent.SchemaRegistry.ISchemaRegistryClient, Confluent.SchemaRegistry.Serdes.AvroSerializerConfig)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.CreateConstructorCallSite(ResultCache lifetime, Type serviceType, Type implementationType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.TryCreateOpenGeneric(ServiceDescriptor descriptor, Type serviceType, CallSiteChain callSiteChain, Int32 slot, Boolean throwOnConstraintViolation)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.TryCreateOpenGeneric(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.CreateCallSite(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.<>c__DisplayClass7_0.<GetCallSite>b__0(Type type)
   at System.Collections.Concurrent.ConcurrentDictionary`2.GetOrAdd(TKey key, Func`2 valueFactory)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.GetCallSite(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.CreateArgumentCallSites(Type serviceType, Type implementationType, CallSiteChain callSiteChain, ParameterInfo[] parameters, Boolean throwIfCallSiteNotFound)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.CreateConstructorCallSite(ResultCache lifetime, Type serviceType, Type implementationType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.TryCreateOpenGeneric(ServiceDescriptor descriptor, Type serviceType, CallSiteChain callSiteChain, Int32 slot, Boolean throwOnConstraintViolation)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.TryCreateOpenGeneric(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.CreateCallSite(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.<>c__DisplayClass7_0.<GetCallSite>b__0(Type type)
   at System.Collections.Concurrent.ConcurrentDictionary`2.GetOrAdd(TKey key, Func`2 valueFactory)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.GetCallSite(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.CreateArgumentCallSites(Type serviceType, Type implementationType, CallSiteChain callSiteChain, ParameterInfo[] parameters, Boolean throwIfCallSiteNotFound)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.CreateConstructorCallSite(ResultCache lifetime, Type serviceType, Type implementationType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.TryCreateOpenGeneric(ServiceDescriptor descriptor, Type serviceType, CallSiteChain callSiteChain, Int32 slot, Boolean throwOnConstraintViolation)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.TryCreateOpenGeneric(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.CreateCallSite(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.<>c__DisplayClass7_0.<GetCallSite>b__0(Type type)
   at System.Collections.Concurrent.ConcurrentDictionary`2.GetOrAdd(TKey key, Func`2 valueFactory)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.CallSiteFactory.GetCallSite(Type serviceType, CallSiteChain callSiteChain)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.ServiceProviderEngine.CreateServiceAccessor(Type serviceType)
   at System.Collections.Concurrent.ConcurrentDictionary`2.GetOrAdd(TKey key, Func`2 valueFactory)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.ServiceProviderEngine.GetService(Type serviceType, ServiceProviderEngineScope serviceProviderEngineScope)
   at Microsoft.Extensions.DependencyInjection.ServiceLookup.ServiceProviderEngineScope.GetService(Type serviceType)
   at Microsoft.Extensions.DependencyInjection.ActivatorUtilities.GetService(IServiceProvider sp, Type type, Type requiredBy, Boolean isDefaultParameterRequired)
   at lambda_method12(Closure , IServiceProvider , Object[] )
   at Microsoft.AspNetCore.Mvc.Controllers.ControllerActivatorProvider.<>c__DisplayClass4_0.<CreateActivator>b__0(ControllerContext controllerContext)
   at Microsoft.AspNetCore.Mvc.Controllers.ControllerFactoryProvider.<>c__DisplayClass5_0.<CreateControllerFactory>g__CreateController|0(ControllerContext controllerContext)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.Next(State& next, Scope& scope, Object& state, Boolean& isCompleted)
   at Microsoft.AspNetCore.Mvc.Infrastructure.ControllerActionInvoker.InvokeInnerFilterAsync()
kmcclellan commented 3 years ago

Hi, thanks for your question! I've also encountered this issue. The type AvroSerializer has problems being resolved by the container because of the constructors it has defined. The error is saying it can't decide whether to use AvroSerializer(ISchemaRegistryClient schemaRegistryClient, IEnumerable<KeyValuePair<string, string>> config) or AvroSerializer(ISchemaRegistryClient schemaRegistryClient, AvroSerializerConfig config = null).

Personally, I consider it a bug that Activator can't choose the same one the compiler chooses when you call new AvroSerializer(client, null). I know of two workarounds, however...

If you don't mind registering each closed generic type, you can use an implementation factory:

services.AddSingleton<IAsyncSerializer<MyType1>>(
    sp => new AvroSerializer<MyType1>(sp.GetRequiredService<ISchemaRegistryClient>()));
services.AddSingleton<IAsyncSerializer<MyType2>>(
    sp => new AvroSerializer<MyType2>(sp.GetRequiredService<ISchemaRegistryClient>()));

Alternatively, you can create a passthrough class with only a single constructor:

class DIAvroSerializer<T> : AvroSerializer<T>
{
    public DIAvroSerializer(ISchemaRegistryClient client, AvroSerializerConfig config = null) : base(client, config) { }
}

Then the open generic registration should work:

services.AddSingleton(typeof(IAsyncSerializer<>), typeof(DIAvroSerializer<>));