getUnregisteredListenerContainer return null

[aliciz]

version: 3.1.1

I am encountering an issue where the getUnregisteredListenerContainer method from the KafkaListenerEndpointRegistry returns null. This occurs in the scenario where I am not using the @KafkaListener annotation but instead, I am manually registering custom containers.

public void pause(String id) {
        MessageListenerContainer listenerContainer = registry.getUnregisteredListenerContainer(id);
        if (listenerContainer == null) {
            log.warn("listenerContainer not exists id:{}", id);
            return;
        }
        listenerContainer.pause();
}

[artembilan]

I am manually registering custom containers.

How do you do that?

The logic there in the refreshContextContainers() is like this:

this.applicationContext.getBeansOfType(MessageListenerContainer.class, true, false).values()
				.forEach(container -> this.unregisteredContainers.put(container.getListenerId(), container));

So, if there is a bean of MessageListenerContainer with that id, it is going to be added to this unregisteredContainers on demand.

[aliciz]

container define like this:

    @Bean(name = KafkaBeanName.DEVICE_DATA_LISTENER_CONTAINER)
    public ConcurrentMessageListenerContainer<String, DeviceData> deviceDataListenerContainer(@Qualifier(KafkaBeanName.DEFAULT_LISTENER_CONTAINER_FACTORY) ConcurrentKafkaListenerContainerFactory<String, DeviceData> factory, IDeviceDataListener deviceDataListener) {
        String topic = KafkaTopic.getDeviceDataTopicByAppId(ttliotSdkProperties.getAppId());
        log.info("afka container init topic:{}", topic);
        ConcurrentMessageListenerContainer<String, DeviceData> container = factory.createContainer(topic);
        Properties props = container.getContainerProperties().getKafkaConsumerProperties();
        props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ListenerContainerInterceptor.class.getName());
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, DeviceData.class.getName());
        container.setupMessageListener(deviceDataListener);
        container.setRecordInterceptor(new DeviceDataRecordInterceptor());
        return container;
    }

You are correct; the unregisteredContainers list does contain the container, but getUnregisteredListenerContainer still returns null.

Can't it be written like this?

@Override
@Nullable
public MessageListenerContainer getUnregisteredListenerContainer(String id) {
  MessageListenerContainer container = this.unregisteredContainers.get(id);
  if (container == null) {
	  refreshContextContainers();
	  return this.unregisteredContainers.get(id);
  }
  return container;
}

[artembilan]

Oh! I see. That is definitely bug 😄
Since that refreshContextContainers(); is called only from this method (and getAllListenerContainers()), it indeed returns null on the subsequent calls for getUnregisteredListenerContainer().

Will fix shortly.
Unfortunately, the release was a couple days ago, so the next one in a month.
As a workaround I suggest to use getAllListenerContainers() and filter that result for desired MessageListenerContainer.