spring-boot - Azure - 使用 AMQP Web 套接字绑定 EventHub 消费者

Question

我正在尝试建立一个非常基本的概念证明，以将 Spring 应用程序链接到 Azure 事件中心。我在启动时不断收到此日志：

[2021-11-17 16:49:18.343 ERROR 25816 --- [           main] o.s.cloud.stream.binding.BindingService : Failed to create consumer binding; retrying in 30 seconds    
...
Caused by: java.lang.IllegalArgumentException: Cannot use a proxy when TransportType is not AMQP Web Sockets.
    at com.azure.messaging.eventhubs.EventHubClientBuilder.getConnectionOptions(EventHubClientBuilder.java:764) ~[azure-messaging-eventhubs-5.10.2.jar:5.10.2]
    at com.azure.messaging.eventhubs.EventHubClientBuilder.buildConnectionProcessor(EventHubClientBuilder.java:705) ~[azure-messaging-eventhubs-5.10.2.jar:5.10.2]
    at com.azure.messaging.eventhubs.EventHubClientBuilder.buildAsyncClient(EventHubClientBuilder.java:638) ~[azure-messaging-eventhubs-5.10.2.jar:5.10.2]
    at com.azure.messaging.eventhubs.EventProcessorClient.<init>(EventProcessorClient.java:89) ~[azure-messaging-eventhubs-5.10.2.jar:5.10.2]
    at com.azure.messaging.eventhubs.EventProcessorClientBuilder.buildEventProcessorClient(EventProcessorClientBuilder.java:585) ~[azure-messaging-eventhubs-5.10.2.jar:5.10.2]
...

尽管我为 EventHubProducerAsyncClient、EventHubConsumerAsyncClient 和 EventProcessorClient 定义了明确地将传输类型设置为AMQP_WEB_SOCKETS. 看这里：

制片人：

@Bean
    public EventHubProducerAsyncClient eventHubProducerAsyncClient() {
         return new EventHubClientBuilder()
                 .connectionString(connectionString)
                 .proxyOptions(new ProxyOptions(
                            ProxyAuthenticationType.NONE,
                            new Proxy(Proxy.Type.HTTP, new InetSocketAddress("my proxy address", 8080)),
                            null, null
                        ))
                 .transportType(AmqpTransportType.AMQP_WEB_SOCKETS)
                 .buildAsyncProducerClient();
    }

消费者：

@Bean
    public EventHubConsumerAsyncClient eventHubConsumerAsyncClient() {
        return new EventHubClientBuilder()
                .connectionString(connectionString)
                .proxyOptions(new ProxyOptions(
                        ProxyAuthenticationType.NONE,
                        new Proxy(Proxy.Type.HTTP, new InetSocketAddress("my proxy address", 8080)),
                        null, null
                    ))
                .transportType(AmqpTransportType.AMQP_WEB_SOCKETS)
                .consumerGroup(DEFAULT_CONSUMER_GROUP_NAME)
                .buildAsyncConsumerClient();
    }

事件处理器：

    private static void onEvent(EventContext eventContext) {
        PartitionContext partition = eventContext.getPartitionContext();
        LOGGER.info("Received events from partition: " + partition.getPartitionId());
    
        EventData event = eventContext.getEventData();
        LOGGER.info("Sequence number: " + event.getSequenceNumber());
        LOGGER.info("Contents: " + new String(event.getBody(), StandardCharsets.UTF_8));
    }
    
    @Bean
    public EventProcessorClient eventProcessorClient() {
        
        BlobContainerAsyncClient blobClient = new BlobContainerClientBuilder()
                .connectionString(storageConnectionString)
                .containerName(storageContainerName)
                .buildAsyncClient();
                
        return new EventProcessorClientBuilder()
                .connectionString(connectionString)
                
                .consumerGroup(DEFAULT_CONSUMER_GROUP_NAME)
                .checkpointStore(new BlobCheckpointStore(blobClient))
                .processEvent(eventContext -> onEvent(eventContext))
                .processError(context -> {
                    LOGGER.error("Error occurred on partition: %s. Error: %s%n",
                            context.getPartitionContext().getPartitionId(), context.getThrowable());
                })
                
                .processPartitionInitialization(initializationContext -> {
                    LOGGER.info("Started receiving on partition: %s%n",
                            initializationContext.getPartitionContext().getPartitionId());
                })
                .processPartitionClose(closeContext -> {
                    LOGGER.info("Stopped receiving on partition: %s. Reason: %s%n",
                            closeContext.getPartitionContext().getPartitionId(),
                            closeContext.getCloseReason());
                })
                .proxyOptions(new ProxyOptions(
                        ProxyAuthenticationType.NONE,
                        new Proxy(Proxy.Type.HTTP, new InetSocketAddress("my proxy address", 8080)),
                        null, null
                    ))
                .transportType(AmqpTransportType.AMQP_WEB_SOCKETS)
                .buildEventProcessorClient();
    }

我在这里遗漏了一些明显的东西吗？我不明白它如何在明确定义时抱怨 TransportType 不是 AMQP Web Sockets。