If Kafka is down, the default kafka connector will try for around 10 times, to ensure the cluster is up. It does this by attempting to get a non-null list of available brokers via an AdminClient.
There are docs on this behaviour at: https://github.com/odpi/egeria/tree/master/open-metadata-implementation/adapters/open-connectors/event-bus-connectors/open-metadata-topic-connectors/kafka-open-metadata-topic-connector
These docs need to be incorporated in the new user docs on egeria-docs
The default also appears to be 60s, not 5s as the doc above states (at least in the coco environment, and I cannot see an override) ie
on Sep 13 12:46:57 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0015 The local server is attempting to connect to Kafka, attempt 2
Mon Sep 13 12:47:57 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0015 The local server is attempting to connect to Kafka, attempt 3
Mon Sep 13 12:48:57 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0015 The local server is attempting to connect to Kafka, attempt 4
Some servers are also members of multiple cohorts. In this case the retry loop is repeated per cohort. For example cocoMDS2 tries cocoCohort then devCohort. After cocoCohort fails, we report an exception saying initialization failed. However the code continues serially with the next cohort (this is all whilst the /instance server start operation is being executed.. and which is in danger of timing out at the http request level)
Mon Sep 13 12:45:56 BST 2021 cocoMDS2 Exception OMRS-AUDIT-0006 Configuration error detected while connecting to cohort devCohort, exception org.odpi.openmetadata.frameworks.connectors.ffdc.ConnectorCheckedException was caught with error message: OCF-KAFKA-TOPIC-CONNECTOR-400-002 Egeria was unable to initialize a connection to a Kafka cluster. The message in the exception was: java.util.concurrent.ExecutionException
Mon Sep 13 12:45:56 BST 2021 cocoMDS2 Exception OMRS-AUDIT-0006 Supplementary information: log record id cb3bac6d-6eff-4661-b750-c70394fa0d76 org.odpi.openmetadata.frameworks.connectors.ffdc.ConnectorCheckedException returned message of OCF-KAFKA-TOPIC-CONNECTOR-400-002 Egeria was unable to initialize a connection to a Kafka cluster. The message in the exception was: java.util.concurrent.ExecutionException and stacktrace of
OCFCheckedExceptionBase{reportedHTTPCode=400, reportingClassName='org.odpi.openmetadata.adapters.eventbus.topic.kafka.KafkaOpenMetadataTopicConnector', reportingActionDescription='KafkaMonitor.waitForThisBroker', reportedErrorMessage='OCF-KAFKA-TOPIC-CONNECTOR-400-002 Egeria was unable to initialize a connection to a Kafka cluster. The message in the exception was: java.util.concurrent.ExecutionException', reportedErrorMessageId='OCF-KAFKA-TOPIC-CONNECTOR-400-002 ', reportedErrorMessageParameters=[java.util.concurrent.ExecutionException, egeria.omag.openmetadata.repositoryservices.cohort.devCohort.OMRSTopic.registration, org.apache.kafka.common.errors.TimeoutException: Call(callName=listNodes, deadlineMs=1631533556915, tries=1, nextAllowedTryMs=1631533557016) timed out at 1631533556916 after 1 attempt(s)], reportedSystemAction='The system is unable initialize.', reportedUserAction='Ensure that Kafka is available', reportedCaughtException=java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=listNodes, deadlineMs=1631533556915, tries=1, nextAllowedTryMs=1631533557016) timed out at 1631533556916 after 1 attempt(s), reportedCaughtExceptionClassName='java.util.concurrent.ExecutionException', relatedProperties=null}
at org.odpi.openmetadata.adapters.eventbus.topic.kafka.KafkaOpenMetadataTopicConnector.start(KafkaOpenMetadataTopicConnector.java:290)
at org.odpi.openmetadata.repositoryservices.connectors.omrstopic.OMRSTopicConnector.start(OMRSTopicConnector.java:258)
at org.odpi.openmetadata.repositoryservices.metadatahighway.OMRSCohortManager.setSecurityVerifier(OMRSCohortManager.java:347)
at org.odpi.openmetadata.repositoryservices.metadatahighway.OMRSMetadataHighwayManager.setSecurityVerifier(OMRSMetadataHighwayManager.java:128)
at org.odpi.openmetadata.repositoryservices.admin.OMRSOperationalServices.setSecurityVerifier(OMRSOperationalServices.java:850)
at org.odpi.openmetadata.adapters.eventbus.topic.kafka.KafkaOpenMetadataTopicConnector$KafkaStatusChecker.waitForBrokers(KafkaOpenMetadataTopicConnector.java:502)
at org.odpi.openmetadata.adapters.eventbus.topic.kafka.KafkaOpenMetadataTopicConnector.start(KafkaOpenMetadataTopicConnector.java:272)
... 61 more
Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=listNodes, deadlineMs=1631533556915, tries=1, nextAllowedTryMs=1631533557016) timed out at 1631533556916 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: listNodes
Mon Sep 13 12:45:56 BST 2021 cocoMDS2 Cohort OMRS-AUDIT-0060 Registering with open metadata repository cohort devCohort using metadata collection id 9614d9b7-d223-4a8c-926d-23ce286c477c
Mon Sep 13 12:45:57 BST 2021 cocoMDS2 Cohort OMRS-AUDIT-0062 Requesting registration information from other members of the open metadata repository cohort devCohort
Mon Sep 13 12:45:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0031 The iotCohort cohort inbound event manager is starting with 2 type definition event consumer(s) and 2 instance event consumer(s)
Mon Sep 13 12:45:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0026 Initializing listener for cohort iotCohort
Mon Sep 13 12:45:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0019 An OMRS Topic Connector has registered with an event bus connector for topic egeria.omag.openmetadata.repositoryservices.cohort.iotCohort.OMRSTopic.registration
Mon Sep 13 12:45:57 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0001 Connecting to Apache Kafka Topic egeria.omag.openmetadata.repositoryservices.cohort.iotCohort.OMRSTopic.registration with a server identifier of 89af21fe-e3fa-4dfe-a307-bfe782616006
Mon Sep 13 12:45:57 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0015 The local server is attempting to connect to Kafka, attempt 1
[I 2021-09-13 12:46:05.873 ServerApp] Saving file at /common/environment-check.ipynb
Mon Sep 13 12:46:57 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0015 The local server is attempting to connect to Kafka, attempt 2
The retry loop in logging an audit message only reports:
Mon Sep 13 12:34:56 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0015 The local server is attempting to connect to Kafka, attempt 10
It may be useful to enhance this with
- what part of server initialization - cohort, which cohort - or topic?
- How many retries in total and/or interval
As this will help someone monitoring the system and understanding what is going wrong.
cocoMDS2 continues to do the same with iotCohort
After cohort initialization is complete we continue with OMASs ie:
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0040 An enterprise OMRS connector has been created for the Connected Asset Services
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0041 The enterprise OMRS connector for the Connected Asset Services has started
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OCF-METADATA-MANAGEMENT-0001 The Open Connector Framework (OCF) Metadata Management Service is initializing a new server instance
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OCF-METADATA-MANAGEMENT-0003 The Open Connector Framework (OCF) Metadata Management Service has initialized a new instance for server cocoMDS2
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMAG-ADMIN-0010 The Open Metadata Access Services (OMASs) are starting
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0040 An enterprise OMRS connector has been created for the Asset Catalog OMAS
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0041 The enterprise OMRS connector for the Asset Catalog OMAS has started
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMAS-ASSET-CATALOG-0002 The Asset Catalog Open Metadata Access Service (OMAS) is initializing a new server instance
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMAS-ASSET-CATALOG-0001 The Asset Catalog Open Metadata Access Service (OMAS) has initialized a new instance for server cocoMDS2
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0040 An enterprise OMRS connector has been created for the Asset Consumer OMAS
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMRS-AUDIT-0041 The enterprise OMRS connector for the Asset Consumer OMAS has started
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OMAS-ASSET-CONSUMER-0001 The Asset Consumer Open Metadata Access Service (OMAS) is initializing a new server instance
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0001 Connecting to Apache Kafka Topic egeria.omag.server.cocoMDS2.omas.assetconsumer.outTopic with a server identifier of 89af21fe-e3fa-4dfe-a307-bfe782616006
Mon Sep 13 12:55:57 BST 2021 cocoMDS2 Startup OCF-KAFKA-TOPIC-CONNECTOR-0015 The local server is attempting to connect to Kafka, attempt 1
After the failure of asset catalog shutdown then starts:
Mon Sep 13 13:05:58 BST 2021 cocoMDS2 Shutdown OMAG-ADMIN-0005 The cocoMDS2 server has begun the shutdown process
So currently the behaviour is
- Try and start all cohorts - do not immediately fail
- Try and start all OMASs - immediately fail
The intent needs clarification and documentation.
- Do we stop after the first timeout is hit - I think this is most likely correct. Kafka not there - we exit after configured retries.
- Do we continue to go through each cohort? ie to allow cohort init to continue in case kafka comes up later - this could be risky, though it's a possibility. For this approach moving the retry loop right into the kafka topic connector consumer/producer threads with a quieter logging approach would simplify the overall logic (see also below re: topics)
- We continue with the existing design - but need to clearly document
It's worth noting that broker availability is one possible check, but there are other failures. For example a topic may not exist - this may be short lived as it will get created by producers, but is still something that could continue indefinately for some topics. Further, if auto topic creation is disabled it will never exist without admin action. This is already handled within the producer/consumer directly and events are already queued to be sent.