Kafka consumer can not be turned off

Current behavior
 There is a kafka consumer and producer introduced, which tries to make connection towards kafka.
Currently we do not support async calls, and we do not necessarily have kafka in our test systems.

If I set localhost:9092 as bootstrap server, than the logs are full with the following messages (logged in each half seconds):

{"logTimeStamp":"2022-06-13T05:05:07.580Z","logLevel":"WARN","principalId":"cpsuser","serviceName":"cps-application","message":"[Consumer clientId=consumer-ncmp-group-1, groupId=ncmp-group] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected","processId":"1","threadName":"org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1","class":"org.apache.kafka.clients.NetworkClient"}

{"logTimeStamp":"2022-06-13T05:05:08.435Z","logLevel":"WARN","principalId":"cpsuser","serviceName":"cps-application","message":"[Consumer clientId=consumer-ncmp-group-1, groupId=ncmp-group] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.","processId":"1","threadName":"org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1","class":"org.apache.kafka.clients.NetworkClient"}

If I set a bootstrap server which is not reachable, NCMP can not startup at all:

 

Expected behavior:

It would be good to be able to turn off both kafka consumer and producer using environment variables

Notification sending (setting up kafka producer) is configurable:
notification:
    async:
        enabled: false

 

However it could come from environment variable too:
notification:
    async:
        enabled: ${NOTIFICATION_ENABLED:false}

 

And it is used at producerConfig creation
@ConditionalOnProperty(name = "notification.async.enabled", havingValue = "true", matchIfMissing = false)

 

I think there should be similar configuration in kafka consumer configuration, to be able to turn off kafka operations if we do not have possibility for async operations at all:
spring:
    kafka:
        enabled: ${KAFKA_ENABLED:false}

And using conditional property at consumer config
@ConditionalOnProperty(name = "spring.kafka.enabled", havingValue = "true", matchIfMissing = false)

 

Reproduction

Set spring.kafka.bootstrap-servers to localhost:9092
That will result error printouts but the service will be able to start up.

 

Set spring.kafka.bootstrap-servers to non reachable external address nonreachable:9092
Service won`t be able to start up.

Try the same with  notification.async.enabled=false, will see the same, because it has no effect on kafka consumer.

 

CPS version

https://gerrit.onap.org/r/gitweb?p=cps.git;a=commit;h=1410509e33142c0c79ff0e111c63abc47f5936d3

 

 

Collected logs

spring.kafka.bootstrap-servers=localhost:9092

ncmp-bootstrap-localhost.log

 

spring.kafka.bootstrap-servers=unreachablehost:9092

ncmp-bootstrap-unreachable.log