Repository: odpi/egeria
Issue: [BUG] Problem registering governance engine / Asset consumer exceptions

### Is there an existing issue for this?

- [X] I have searched the existing issues

### Current Behavior

Noticed this when checking the notebooks for release.

This notebook is explicitly marked as not to be run. However it looks to be an issue worth checking, possibly after some refactoring

```
 
POST https://lab-core:9443/servers/cocoMDS2/open-metadata/access-services/governance-engine/users/erinoverview/governance-services/new/GovernanceActionService
{
    "class": "NewGovernanceServiceRequestBody",
    "qualifiedName": "ftp-governance-action-service",
    "displayName": "FTP Governance Action Service",
    "description": "Simulates FTP from an external party.",
    "connection": {
        "class": "Connection",
        "type": {
            "class": "ElementType",
            "elementTypeId": "114e9f8f-5ff3-4c32-bd37-a7eb42712253",
            "elementTypeName": "Connection",
            "elementTypeVersion": 1,
            "elementTypeDescription": "A set of properties to identify and configure a connector instance.",
            "elementOrigin": "CONFIGURATION"
        },
        "qualifiedName": "ftp-governance-action-service-implementation",
        "displayName": "FTP Governance Action Service Implementation Connection",
        "description": "Connection for governance service ftp-governance-action-service",
        "connectorType": {
            "class": "ConnectorType",
            "type": {
                "class": "ElementType",
                "elementTypeId": "954421eb-33a6-462d-a8ca-b5709a1bd0d4",
                "elementTypeName": "ConnectorType",
                "elementTypeVersion": 1,
                "elementTypeDescription": "A set of properties describing a type of connector.",
                "elementOrigin": "LOCAL_COHORT"
            },
            "guid": "1111f73d-e343-abcd-82cb-3918fed81da6",
            "qualifiedName": "ftp-governance-action-service-GovernanceServiceProvider",
            "displayName": "FTP Governance Action Service Governance Service Provider Implementation",
            "description": "Simulates FTP from an external party.",
            "connectorProviderClassName": "org.odpi.openmetadata.adapters.connectors.governanceactions.provisioning.MoveCopyFileGovernanceActionProvider"
        },
        "configurationProperties": {
            "noLineage": ""
        }
    }
}
 
Returns:
{
    "class": "GUIDResponse",
    "relatedHTTPCode": 200,
    "guid": "d594dc4d-59f6-44c8-8121-0dafdc5be544"
}
 
 
The guid for the ftp-governance-action-service governance service is: d594dc4d-59f6-44c8-8121-0dafdc5be544
 
POST https://lab-datalake:9443/servers/cocoMDS1/open-metadata/access-services/governance-engine/users/peterprofile/governance-engines/f45b9bb7-81d7-4304-a74f-b8162a3438e9/governance-services
{
    "class": "GovernanceServiceRegistrationRequestBody",
    "governanceServiceGUID": "d594dc4d-59f6-44c8-8121-0dafdc5be544",
    "requestType": "copy-file"
}
 
Returns:
{
    "class": "VoidResponse",
    "relatedHTTPCode": 500,
    "exceptionClassName": "org.odpi.openmetadata.frameworks.connectors.ffdc.PropertyServerException",
    "actionDescription": "registerGovernanceServiceWithEngine",
    "exceptionErrorMessage": "OMAG-REPOSITORY-HANDLER-500-001 An unexpected error org.odpi.openmetadata.repositoryservices.ffdc.exception.RepositoryErrorException was returned to reclassifyEntity(LatestChange) by the metadata server during registerGovernanceServiceWithEngine request for open metadata access service Governance Engine OMAS on server cocoMDS1; message was OMRS-REPOSITORY-400-056 The OMRS repository connector operation updateEntityClassification (EntityProxy) from the OMRS Enterprise Repository Services can not locate the home repository connector for instance LatestChange located in metadata collection 94645157-2a9d-4d32-9170-e6f2747a0ceb",
    "exceptionErrorMessageId": "OMAG-REPOSITORY-HANDLER-500-001",
    "exceptionErrorMessageParameters": [
        "OMRS-REPOSITORY-400-056 The OMRS repository connector operation updateEntityClassification (EntityProxy) from the OMRS Enterprise Repository Services can not locate the home repository connector for instance LatestChange located in metadata collection 94645157-2a9d-4d32-9170-e6f2747a0ceb",
        "registerGovernanceServiceWithEngine",
        "Governance Engine OMAS",
        "cocoMDS1",
        "org.odpi.openmetadata.repositoryservices.ffdc.exception.RepositoryErrorException",
        "reclassifyEntity(LatestChange)"
    ],
    "exceptionSystemAction": "The system is unable to process the request because of an internal error.",
    "exceptionUserAction": "Verify the sanity of the server.  This is probably a logic error.  If you can not work out what happened, ask the Egeria community for help."
}
 
OMAG-REPOSITORY-HANDLER-500-001 An unexpected error org.odpi.openmetadata.repositoryservices.ffdc.exception.RepositoryErrorException was returned to reclassifyEntity(LatestChange) by the metadata server during registerGovernanceServiceWithEngine request for open metadata access service Governance Engine OMAS on server cocoMDS1; message was OMRS-REPOSITORY-400-056 The OMRS repository connector operation updateEntityClassification (EntityProxy) from the OMRS Enterprise Repository Services can not locate the home repository connector for instance LatestChange located in metadata collection 94645157-2a9d-4d32-9170-e6f2747a0ceb
 * The system is unable to process the request because of an internal error.
 * Verify the sanity of the server.  This is probably a logic error.  If you can not work out what happened, ask the Egeria community for help.
Service registered as: copy-file
```

### Expected Behavior

governance engine registration passes

### Steps To Reproduce

Run the incomplete/unsupported notebook 'automated curation' in a coco pharma tutorial environment

### Environment

```markdown
- Egeria:3.8
```


### Any Further Information?

_No response_

Changed files:
open-metadata-implementation/adapters/open-connectors/event-bus-connectors/open-metadata-topic-connectors/kafka-open-metadata-topic-connector/src/main/java/org/odpi/openmetadata/adapters/eventbus/topic/kafka/KafkaOpenMetadataEventConsumer.java

diff --git a/open-metadata-implementation/adapters/open-connectors/event-bus-connectors/open-metadata-topic-connectors/kafka-open-metadata-topic-connector/src/main/java/org/odpi/openmetadata/adapters/eventbus/topic/kafka/KafkaOpenMetadataEventConsumer.java b/open-metadata-implementation/adapters/open-connectors/event-bus-connectors/open-metadata-topic-connectors/kafka-open-metadata-topic-connector/src/main/java/org/odpi/openmetadata/adapters/eventbus/topic/kafka/KafkaOpenMetadataEventConsumer.java
index 4c68da4943..50232bb6bd 100644
--- a/open-metadata-implementation/adapters/open-connectors/event-bus-connectors/open-metadata-topic-connectors/kafka-open-metadata-topic-connector/src/main/java/org/odpi/openmetadata/adapters/eventbus/topic/kafka/KafkaOpenMetadataEventConsumer.java
+++ b/open-metadata-implementation/adapters/open-connectors/event-bus-connectors/open-metadata-topic-connectors/kafka-open-metadata-topic-connector/src/main/java/org/odpi/openmetadata/adapters/eventbus/topic/kafka/KafkaOpenMetadataEventConsumer.java
@@ -3,30 +3,21 @@
 package org.odpi.openmetadata.adapters.eventbus.topic.kafka;
 
 import java.time.Duration;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Queue;
+import java.util.*;
 import java.util.concurrent.BlockingDeque;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.kafka.clients.consumer.CommitFailedException;
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.*;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 import org.odpi.openmetadata.frameworks.auditlog.AuditLog;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static java.lang.Thread.sleep;
 
 
 /**
@@ -52,7 +43,10 @@ public class KafkaOpenMetadataEventConsumer implements Runnable
     private long nextMessageProcessingStatusCheckTime = System.currentTimeMillis();
     private long maxNextPollTimestampToAvoidConsumerTimeout = 0;
     private final long maxMsBetweenPolls;
-    
+
+    // Keep track of when an initial rebalance is done
+    private boolean initialPartitionAssignment = true;
+
     
     //If we get close enough to the consumer timeout timestamp, force a poll so that
     //we do not exceed the timeout.  This parameter controls how close we can get
@@ -68,6 +62,7 @@ public class KafkaOpenMetadataEventConsumer implements Runnable
     private final AtomicBoolean running = new AtomicBoolean(true);
     
     private final boolean isAutoCommitEnabled;
+    private final long startTime = System.currentTimeMillis();
 
     /**
      * Constructor for the event consumer.
@@ -112,6 +107,7 @@ public class KafkaOpenMetadataEventConsumer implements Runnable
         this.messageProcessingStatusCheckIntervalMs = config.getLongProperty(KafkaOpenMetadataEventConsumerProperty.COMMIT_CHECK_INTERVAL_MS);
         long messageTimeoutMins = config.getLongProperty(KafkaOpenMetadataEventConsumerProperty.CONSUMER_EVENT_PROCESSING_TIMEOUT_MINS);
         this.messageProcessingTimeoutMs = messageTimeoutMins < 0 ? messageTimeoutMins : TimeUnit.MILLISECONDS.convert(messageTimeoutMins, TimeUnit.MINUTES);
+
     }
 
 
@@ -135,8 +131,7 @@ public class KafkaOpenMetadataEventConsumer implements Runnable
     public void run()
     {
         final String           actionDescription = "run";
-        KafkaOpenMetadataTopicConnectorAuditCode auditCode;
-       
+
         while (isRunning())
         {
             try
@@ -452,7 +447,7 @@ public class KafkaOpenMetadataEventConsumer implements Runnable
     private void awaitNextPollingTime() {
 		try
 		{
-		    Thread.sleep(1000);
+		    sleep(1000);
 		}
 		catch (InterruptedException e)
 		{
@@ -468,7 +463,7 @@ public class KafkaOpenMetadataEventConsumer implements Runnable
 
         try
         {
-            Thread.sleep(recoverySleepTimeSec * 1000L);
+            sleep(recoverySleepTimeSec * 1000L);
         }
         catch (InterruptedException e1)
         {
@@ -514,17 +509,68 @@ public class KafkaOpenMetadataEventConsumer implements Runnable
     }
 
 
-    private class HandleRebalance implements ConsumerRebalanceListener
-    {
+    private class HandleRebalance implements ConsumerRebalanceListener {
         AuditLog auditLog = null;
+
         public HandleRebalance(AuditLog auditLog) {
             this.auditLog = auditLog;
         }
 
-        public void onPartitionsAssigned(Collection<TopicPartition> partitions)
-        {
+        @Override
+        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+
+            // Check if we need to rewind to handle initial startup case -- but only on first assignment
+            try {
+                if (initialPartitionAssignment) {
+                    log.info("Received initial PartitionsAssigned event");
+
+                    long partitionCount = partitions.size();
+
+                    if (partitionCount != 1) {
+                        log.info("Received PartitionsAssigned event with {} partitions. This is not supported.",partitionCount);
+                    } else {
+                        // there is only one partition, so we can just grab the first one - and we'll try this once only
+                        initialPartitionAssignment = false;
+                        long maxOffsetWanted = 0; // same as 'beginning'
+
+                        TopicPartition partition = partitions.iterator().next();
+
+                        // query offset by timestamp (when we started connector) - NULL if there are no messages later than this offset
+                        long reqStartTime=KafkaOpenMetadataEventConsumer.this.startTime;
+                        log.info("Querying for offset by timestamp: {}",reqStartTime);
+                        OffsetAndTimestamp otByStartTime = consumer.offsetsForTimes(Collections.singletonMap(partition,
+                                reqStartTime)).get(partition);
+
+                        // If null, then we don't have any earlier messages - ie there is no offset found
+                        if (otByStartTime != null) {
+                            // where we want to scoll to - the messages sent since we thought we started
+                            maxOffsetWanted = otByStartTime.offset();
+                            log.info("Earliest offset found for {} is {}",reqStartTime,otByStartTime.timestamp());
+
+                            // get the current offset
+                            long currentOffset = consumer.position(partition);
+
+                            // if the current offset is later than the start time we want, rewind to the start time
+                            if (currentOffset > maxOffsetWanted) {
+
+                                log.info("Seeking to {} for partition {} and topic {} as current offset {} is too late", maxOffsetWanted, partition.partition(),
+                                        partition.topic(), currentOffset);
+                                consumer.seek(partition, maxOffsetWanted);
+                            } else
+                                log.info("Not Seeking to {} for partition {} and topic {} as current offset {} is older", maxOffsetWanted, partition.partition(),
+                                        partition.topic(), currentOffset);
+                        }
+                        else
+                            log.info("No missed events found for partition {} and topic {}", partition.partition(), partition.topic());
+                    }
+                }
+            } catch (Exception e) {
+                // We leave the offset as-is if anything goes wrong. Eventually other messages will cause the effective state to be updated
+                log.info("Error correcting seek position, continuing with defaults", e);
+            }
         }
 
+        @Override
         public void onPartitionsRevoked(Collection<TopicPartition> partitions)
         {
             final String methodName = "onPartitionsRevoked.commitSync";