Azure eventProcessorclient consumer giving HttpProxyHandler is not a @Sharable handler

[ssudarjothi]

Hi,

I have Azure Event consumer which is consuming the events using EventProcessorClient. I am creating the Credential, and the blob checkpoint store, and then connecting to the consumer using EventProcessorClient. I am using Proxy in all the 3 steps, Credential creation, Blob Checkpoint creation and in the EventProcessorClient, to connect from my client network to the Azure event hub, Its running fine for 10+ hours, i am able to consume the messages, and after that its giving me the below mentioned error. When i wake up in the morning, its having 10GB of error logs with the same message. Please let me know how to resolve this.

I01-07-2020 22:00:53.920 - WARN i.n.c.ChannelInitializer - Failed to initialize a channel. Closing: [id: 0x95041f0f]
io.netty.channel.ChannelPipelineException: com.azure.core.http.netty.implementation.HttpProxyHandler is not a @sharable handler, so can't be added or removed multiple times.
at io.netty.channel.DefaultChannelPipeline.checkMultiplicity(DefaultChannelPipeline.java:625)
at io.netty.channel.DefaultChannelPipeline.addFirst(DefaultChannelPipeline.java:159)
at io.netty.channel.DefaultChannelPipeline.addFirst(DefaultChannelPipeline.java:152)
at com.azure.core.http.netty.NettyAsyncHttpClient.lambda$configureTcpClient$1(NettyAsyncHttpClient.java:120)
at reactor.netty.channel.BootstrapHandlers$BootstrapInitializerHandler.initChannel(BootstrapHandlers.java:455)
at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:115)
at io.netty.channel.ChannelInitializer.handlerAdded(ChannelInitializer.java:107)
at io.netty.channel.DefaultChannelPipeline.callHandlerAdded0(DefaultChannelPipeline.java:637)
at io.netty.channel.DefaultChannelPipeline.addFirst(DefaultChannelPipeline.java:187)
at io.netty.channel.DefaultChannelPipeline.addFirst(DefaultChannelPipeline.java:384)
at io.netty.channel.DefaultChannelPipeline.addFirst(DefaultChannelPipeline.java:363)
at reactor.netty.resources.PooledConnectionProvider$Pool.channelCreated(PooledConnectionProvider.java:313)
at io.netty.channel.pool.SimpleChannelPool$1.initChannel(SimpleChannelPool.java:113)
at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:115)
at io.netty.channel.ChannelInitializer.handlerAdded(ChannelInitializer.java:107)
at io.netty.channel.DefaultChannelPipeline.callHandlerAdded0(DefaultChannelPipeline.java:637)
at io.netty.channel.DefaultChannelPipeline.access$000(DefaultChannelPipeline.java:46)
at io.netty.channel.DefaultChannelPipeline$PendingHandlerAddedTask.execute(DefaultChannelPipeline.java:1487)
at io.netty.channel.DefaultChannelPipeline.callHandlerAddedForAllHandlers(DefaultChannelPipeline.java:1161)
at io.netty.channel.DefaultChannelPipeline.invokeHandlerAddedIfNeeded(DefaultChannelPipeline.java:686)
at io.netty.channel.AbstractChannel$AbstractUnsafe.register0(AbstractChannel.java:510)
at io.netty.channel.AbstractChannel$AbstractUnsafe.register(AbstractChannel.java:476)
at io.netty.channel.SingleThreadEventLoop.register(SingleThreadEventLoop.java:80)
at io.netty.channel.SingleThreadEventLoop.register(SingleThreadEventLoop.java:74)
at reactor.netty.resources.ColocatedEventLoopGroup.register(ColocatedEventLoopGroup.java:69)
at io.netty.bootstrap.AbstractBootstrap.initAndRegister(AbstractBootstrap.java:333)
at io.netty.bootstrap.Bootstrap.doResolveAndConnect(Bootstrap.java:163)
at io.netty.bootstrap.Bootstrap.connect(Bootstrap.java:119)
at io.netty.channel.pool.SimpleChannelPool.connectChannel(SimpleChannelPool.java:266)
at io.netty.channel.pool.SimpleChannelPool.acquireHealthyFromPoolOrNew(SimpleChannelPool.java:179)
at io.netty.channel.pool.SimpleChannelPool.notifyHealthCheck(SimpleChannelPool.java:251)
at io.netty.channel.pool.SimpleChannelPool.doHealthCheck(SimpleChannelPool.java:226)
at io.netty.channel.pool.SimpleChannelPool.access$100(SimpleChannelPool.java:42)
at io.netty.channel.pool.SimpleChannelPool$3.run(SimpleChannelPool.java:199)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404)
at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:335)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:897)
at java.lang.Thread.run(Thread.java:748)

Given below is the code snippet

netSocketAddress inetSocketAddress=new InetSocketAddress(proxy, proxyport);

     com.azure.core.http.ProxyOptions httpProxyOption=new com.azure.core.http.ProxyOptions(com.azure.core.http.ProxyOptions.Type.HTTP,inetSocketAddress);
     httpProxyOption.setCredentials("","");

   TokenCredential credential = new ClientSecretCredentialBuilder()
         .tenantId(tenantId)
         .clientId(clientId)
         .clientSecret(clientSecret)
         .proxyOptions(httpProxyOption)
         .build(); 

BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder()
.endpoint(containerUrl)
.credential(credential)
.httpClient(httpClient)
.buildAsyncClient();

EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder()
.processEvent(processEvent)
.processError(processError)
.processPartitionInitialization(initializePartition)
.processPartitionClose(closePartition)
.credential(fullyQualifiedNamespace, eventHubName, credential)
.consumerGroup(consumerGroup)
.checkpointStore(blobCheckpoint)
.initialPartitionEventPosition(map)
.proxyOptions(getproxyOptions())
.transportType(AmqpTransportType.AMQP_WEB_SOCKETS)
.buildEventProcessorClient();

[srnagar]

@ssudarjothi Thanks for reporting the issue. Can you please include the dependency versions you are using for the following Azure client libraries?

azure-core
azure-core-amqp
azure-storage-blob
azure-messaging-eventhubs
azure-messaging-eventhubs-checkpointstore-blob

[ssudarjothi]

@srnagar Given below are the dependencies i have in my project. I am using azure-sdk-all,instead of azure-core, core-amqp.

    compile('com.azure:azure-sdk-all:1.0.0')
    compile('com.azure:azure-storage-blob:12.6.1')
compile('com.azure:azure-messaging-eventhubs:5.1.1')
compile('com.azure:azure-messaging-eventhubs-checkpointstore-blob:1.1.1')
compile('com.azure:azure-identity:1.0.6')

[alzimmermsft]

Fixed in #13304

I'll post a comment on this issue once the fix has been shipped in a GA version.