Support Kafka Kraft Mode for use in Confluent-local Images

[scprek]

Feature description

Confluent provides a confluent-local

More details: https://docs.confluent.io/platform/current/installation/docker/image-reference.html#ak-images

Although we can replace the image, test containers has a method .kraftMode() that modifies the necessary things to allow using kraft mode. Things like the startup verification.

https://java.testcontainers.org/modules/kafka/#using-kraft-mode

Today if you specify the image, testcontainers will fail to know if it started up healthy or not after a timeout

test-resources:
  containers:
    kafka:
      image-name: confluentinc/confluent-local:latest   

error

Error instantiating bean of type  [io.micronaut.configuration.kafka.config.DefaultKafkaConsumerConfiguration]

Message: Test resources service wasn't able to revolve expression 'kafka.bootstrap.servers': java.net.http.HttpTimeoutException: request timed out
Path Taken: new KafkaConsumerProcessor(ExecutorService executorService,ApplicationConfiguration applicationConfiguration,BeanContext beanContext,AbstractKafkaConsumerConfiguration defaultConsumerConfiguration,ConsumerRecordBinderRegistry binderRegistry,BatchConsumerRecordsBinderRegistry batchBinderRegistry,SerdeRegistry serdeRegistry,ProducerRegistry producerRegistry,KafkaListenerExceptionHandler exceptionHandler,ExecutorService schedulerService,TransactionalProducerRegistry transactionalProducerRegistry,ApplicationEventPublisher startedEventPublisher,ApplicationEventPublisher subscribedEventPublisher,ConditionalRetryBehaviourHandler conditionalRetryBehaviourHandler) --> new KafkaConsumerProcessor(ExecutorService executorService,ApplicationConfiguration applicationConfiguration,BeanContext beanContext,[AbstractKafkaConsumerConfiguration defaultConsumerConfiguration],ConsumerRecordBinderRegistry binderRegistry,BatchConsumerRecordsBinderRegistry batchBinderRegistry,SerdeRegistry serdeRegistry,ProducerRegistry producerRegistry,KafkaListenerExceptionHandler exceptionHandler,ExecutorService schedulerService,TransactionalProducerRegistry transactionalProducerRegistry,ApplicationEventPublisher startedEventPublisher,ApplicationEventPublisher subscribedEventPublisher,ConditionalRetryBehaviourHandler conditionalRetryBehaviourHandler) --> new DefaultKafkaConsumerConfiguration([KafkaDefaultConfiguration defaultConfiguration])
io.micronaut.context.exceptions.BeanInstantiationException: Error instantiating bean of type  [io.micronaut.configuration.kafka.config.DefaultKafkaConsumerConfiguration]

Message: Test resources service wasn't able to revolve expression 'kafka.bootstrap.servers': java.net.http.HttpTimeoutException: request timed out
Path Taken: new KafkaConsumerProcessor(ExecutorService executorService,ApplicationConfiguration applicationConfiguration,BeanContext beanContext,AbstractKafkaConsumerConfiguration defaultConsumerConfiguration,ConsumerRecordBinderRegistry binderRegistry,BatchConsumerRecordsBinderRegistry batchBinderRegistry,SerdeRegistry serdeRegistry,ProducerRegistry producerRegistry,KafkaListenerExceptionHandler exceptionHandler,ExecutorService schedulerService,TransactionalProducerRegistry transactionalProducerRegistry,ApplicationEventPublisher startedEventPublisher,ApplicationEventPublisher subscribedEventPublisher,ConditionalRetryBehaviourHandler conditionalRetryBehaviourHandler) --> new KafkaConsumerProcessor(ExecutorService executorService,ApplicationConfiguration applicationConfiguration,BeanContext beanContext,[AbstractKafkaConsumerConfiguration defaultConsumerConfiguration],ConsumerRecordBinderRegistry binderRegistry,BatchConsumerRecordsBinderRegistry batchBinderRegistry,SerdeRegistry serdeRegistry,ProducerRegistry producerRegistry,KafkaListenerExceptionHandler exceptionHandler,ExecutorService schedulerService,TransactionalProducerRegistry transactionalProducerRegistry,ApplicationEventPublisher startedEventPublisher,ApplicationEventPublisher subscribedEventPublisher,ConditionalRetryBehaviourHandler conditionalRetryBehaviourHandler) --> new DefaultKafkaConsumerConfiguration([KafkaDefaultConfiguration defaultConfiguration])
	at io.micronaut.context.DefaultBeanContext.resolveByBeanFactory(DefaultBeanContext.java:2326)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2281)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2293)
	at io.micronaut.context.DefaultBeanContext.createRegistration(DefaultBeanContext.java:3095)
	at io.micronaut.context.SingletonScope.getOrCreate(SingletonScope.java:80)
	at io.micronaut.context.DefaultBeanContext.findOrCreateSingletonBeanRegistration(DefaultBeanContext.java:2997)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2958)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2732)
	at io.micronaut.context.DefaultBeanContext.getBean(DefaultBeanContext.java:1731)
	at io.micronaut.context.AbstractBeanResolutionContext.getBean(AbstractBeanResolutionContext.java:89)
	at io.micronaut.context.AbstractInitializableBeanDefinition.resolveBean(AbstractInitializableBeanDefinition.java:2161)
	at io.micronaut.context.AbstractInitializableBeanDefinition.getBeanForConstructorArgument(AbstractInitializableBeanDefinition.java:1324)
	at io.micronaut.configuration.kafka.config.$DefaultKafkaConsumerConfiguration$Definition.instantiate(Unknown Source)
	at io.micronaut.context.DefaultBeanContext.resolveByBeanFactory(DefaultBeanContext.java:2311)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2281)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2293)
	at io.micronaut.context.DefaultBeanContext.createRegistration(DefaultBeanContext.java:3095)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2985)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2732)
	at io.micronaut.context.DefaultBeanContext.getBean(DefaultBeanContext.java:1731)
	at io.micronaut.context.AbstractBeanResolutionContext.getBean(AbstractBeanResolutionContext.java:89)
	at io.micronaut.context.AbstractInitializableBeanDefinition.resolveBean(AbstractInitializableBeanDefinition.java:2161)
	at io.micronaut.context.AbstractInitializableBeanDefinition.getBeanForConstructorArgument(AbstractInitializableBeanDefinition.java:1324)
	at io.micronaut.configuration.kafka.processor.$KafkaConsumerProcessor$Definition.instantiate(Unknown Source)
	at io.micronaut.context.DefaultBeanContext.resolveByBeanFactory(DefaultBeanContext.java:2311)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2281)
	at io.micronaut.context.DefaultBeanContext.doCreateBean(DefaultBeanContext.java:2293)
	at io.micronaut.context.DefaultBeanContext.createRegistration(DefaultBeanContext.java:3095)
	at io.micronaut.context.SingletonScope.getOrCreate(SingletonScope.java:80)
	at io.micronaut.context.DefaultBeanContext.findOrCreateSingletonBeanRegistration(DefaultBeanContext.java:2997)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2958)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistration(DefaultBeanContext.java:2932)
	at io.micronaut.context.DefaultBeanContext.addCandidateToList(DefaultBeanContext.java:3521)
	at io.micronaut.context.DefaultBeanContext.resolveBeanRegistrations(DefaultBeanContext.java:3476)
	at io.micronaut.context.DefaultBeanContext.getBeanRegistrations(DefaultBeanContext.java:3450)
	at io.micronaut.context.DefaultBeanContext.streamOfType(DefaultBeanContext.java:980)
	at io.micronaut.context.DefaultBeanContext.streamOfType(DefaultBeanContext.java:906)
	at io.micronaut.context.DefaultBeanContext.streamOfType(DefaultBeanContext.java:883)
	at io.micronaut.context.DefaultBeanContext.initializeContext(DefaultBeanContext.java:2020)
	at io.micronaut.context.DefaultApplicationContext.initializeContext(DefaultApplicationContext.java:294)
	at io.micronaut.context.DefaultBeanContext.readAllBeanDefinitionClasses(DefaultBeanContext.java:3348)
	at io.micronaut.context.DefaultBeanContext.finalizeConfiguration(DefaultBeanContext.java:3692)
	at io.micronaut.context.DefaultBeanContext.start(DefaultBeanContext.java:342)
	at io.micronaut.context.DefaultApplicationContext.start(DefaultApplicationContext.java:202)
	at io.micronaut.test.extensions.AbstractMicronautExtension.startApplicationContext(AbstractMicronautExtension.java:507)
	at io.micronaut.test.extensions.AbstractMicronautExtension.beforeClass(AbstractMicronautExtension.java:346)
	at io.micronaut.test.extensions.junit5.MicronautJunit5Extension.beforeAll(MicronautJunit5Extension.java:84)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
Caused by: io.micronaut.testresources.client.TestResourcesException: Test resources service wasn't able to revolve expression 'kafka.bootstrap.servers': java.net.http.HttpTimeoutException: request timed out
	at io.micronaut.testresources.client.TestResourcesClientPropertyExpressionResolver$DelegateResolver.withErrorHandling(TestResourcesClientPropertyExpressionResolver.java:130)
	at io.micronaut.testresources.client.TestResourcesClientPropertyExpressionResolver$DelegateResolver.callClient(TestResourcesClientPropertyExpressionResolver.java:112)
	at io.micronaut.testresources.client.TestResourcesClientPropertyExpressionResolver$DelegateResolver.resolve(TestResourcesClientPropertyExpressionResolver.java:99)
	at io.micronaut.testresources.core.LazyTestResourcesExpressionResolver.resolve(LazyTestResourcesExpressionResolver.java:45)
	at io.micronaut.context.env.DefaultPropertyPlaceholderResolver.resolveExpression(DefaultPropertyPlaceholderResolver.java:209)
	at io.micronaut.context.env.DefaultPropertyPlaceholderResolver$PlaceholderSegment.getValue(DefaultPropertyPlaceholderResolver.java:381)
	at io.micronaut.context.env.DefaultPropertyPlaceholderResolver.resolveRequiredPlaceholdersObject(DefaultPropertyPlaceholderResolver.java:116)
	at io.micronaut.context.env.PropertySourcePropertyResolver.resolvePlaceHoldersIfNecessary(PropertySourcePropertyResolver.java:791)
	at io.micronaut.context.env.PropertySourcePropertyResolver.resolveSubMap(PropertySourcePropertyResolver.java:561)
	at io.micronaut.context.env.PropertySourcePropertyResolver.getProperties(PropertySourcePropertyResolver.java:293)
	at io.micronaut.core.value.PropertyResolver.getProperties(PropertyResolver.java:113)
	at io.micronaut.configuration.kafka.config.KafkaDefaultConfiguration.resolveDefaultConfiguration(KafkaDefaultConfiguration.java:82)
	at io.micronaut.configuration.kafka.config.KafkaDefaultConfiguration.<init>(KafkaDefaultConfiguration.java:54)
	at io.micronaut.configuration.kafka.config.$KafkaDefaultConfiguration$Definition.instantiate(Unknown Source)
	at io.micronaut.context.DefaultBeanContext.resolveByBeanFactory(DefaultBeanContext.java:2311)
	... 47 more