HDDS-14006. Address Copilot review comments

Also make the seek position string be deserialized from a string to a
long earlier in the process so it doesn't have to be deserialized in
multiple places and the DB list calls can just take a Long as the seek
position

Author: Colm Dougan

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContext.java

@@ -31,7 +31,7 @@ public interface OMEventListenerPluginContext {
 
   // TODO: should we allow plugins to pass in maxResults or just limit
   // them to some predefined value for safety?  e.g. 10K
-  List<OmCompletedRequestInfo> listCompletedRequestInfo(String startKey, int maxResults) throws IOException;
+  List<OmCompletedRequestInfo> listCompletedRequestInfo(Long startKey, int maxResults) throws IOException;
 
   // XXX: this probably doesn't belong here
   String getThreadNamePrefix();

hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java

@@ -354,7 +354,7 @@ List<OmVolumeArgs> listVolumes(String userName, String prefix,
    * @return a list of {@link OmCompletedRequestInfo}
    * @throws IOException
    */
-  List<OmCompletedRequestInfo> listCompletedRequestInfo(String startKey, int maxResults) throws IOException;
+  List<OmCompletedRequestInfo> listCompletedRequestInfo(Long startKey, int maxResults) throws IOException;
 
   /**
    * Returns the names of up to {@code count} open keys whose age is

hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerKafkaPublisher.java

@@ -60,8 +60,8 @@ public void initialize(OzoneConfiguration conf, OMEventListenerPluginContext plu
     long kafkaServiceTimeout = 300 * 1000;
 
     LOG.info("Creating OMEventListenerLedgerPoller with serviceInterval={}," +
-             "serviceTimeout={}, kafkaProps={}, seekPosition={}",
-        kafkaServiceInterval, kafkaServiceTimeout, kafkaProps,
+             "serviceTimeout={}, seekPosition={}",
+        kafkaServiceInterval, kafkaServiceTimeout,
         seekPosition);
 
     this.seekPosition = new OMEventListenerLedgerPollerSeekPosition();
@@ -135,14 +135,16 @@ static class KafkaClientWrapper {
     }
 
     public void initialize() throws IOException {
-      LOG.info("Initializing with properties {}", kafkaProps);
+      LOG.info("Initializing kafka client for topic {}", topic);
       this.producer = new KafkaProducer<>(kafkaProps);
 
       ensureTopicExists();
     }
 
     public void shutdown() throws IOException {
-      producer.close();
+      if (producer != null) {
+        producer.close();
+      }
     }
 
     public void send(String message) throws IOException {

hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPoller.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.utils.BackgroundService;
 import org.apache.hadoop.hdds.utils.BackgroundTask;
@@ -59,7 +60,7 @@ public OMEventListenerLedgerPoller(long interval, TimeUnit unit,
 
     super("OMEventListenerLedgerPoller",
           interval,
-          TimeUnit.MILLISECONDS,
+          unit,
           poolSize,
           serviceTimeout, pluginContext.getThreadNamePrefix());
 
@@ -116,13 +117,15 @@ public BackgroundTaskResult call() {
           LOG.debug("Running OMEventListenerLedgerPoller");
         }
         if (runCount.get() == 0) {
-          seekPosition.initSeekPosition();
+          seekPosition.set(seekPosition.initSeekPosition());
         }
         getRunCount().incrementAndGet();
 
         try {
+          String startKeyStr = seekPosition.get();
+          Long startKey = StringUtils.isNotBlank(startKeyStr) ? Long.valueOf(startKeyStr) : null;
           for (OmCompletedRequestInfo requestInfo : pluginContext.listCompletedRequestInfo(
-                  seekPosition.get(), MAX_RESULTS)) {
+                  startKey, MAX_RESULTS)) {
             callback.accept(requestInfo);
           }
           successRunCount.incrementAndGet();

hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerKafkaPublisher.java

@@ -38,7 +38,7 @@
 import org.mockito.junit.jupiter.MockitoExtension;
 
 /**
- * Tests {@link OMEventListenerPluginManager}.
+ * Tests {@link OMEventListenerKafkaPublisher}.
  */
 @ExtendWith(MockitoExtension.class)
 public class TestOMEventListenerKafkaPublisher {
@@ -85,13 +85,13 @@ private List<String> captureEventsProducedByOperation(OmCompletedRequestInfo op,
     List<String> events = new ArrayList<>();
 
     OMEventListenerKafkaPublisher plugin = new OMEventListenerKafkaPublisher();
-    try (MockedConstruction<OMEventListenerKafkaPublisher.KafkaClientWrapper> mockeKafkaClientWrapper =
+    try (MockedConstruction<OMEventListenerKafkaPublisher.KafkaClientWrapper> mockedKafkaClientWrapper =
              mockConstruction(OMEventListenerKafkaPublisher.KafkaClientWrapper.class)) {
 
       plugin.initialize(conf, pluginContext);
       plugin.handleCompletedRequest(op);
 
-      OMEventListenerKafkaPublisher.KafkaClientWrapper mock = mockeKafkaClientWrapper.constructed().get(0);
+      OMEventListenerKafkaPublisher.KafkaClientWrapper mock = mockedKafkaClientWrapper.constructed().get(0);
       ArgumentCaptor<String> argument = ArgumentCaptor.forClass(String.class);
       verify(mock, times(expectEvents)).send(argument.capture());
 
@@ -144,7 +144,7 @@ public void testCreateDirectoryRequestProducesS3CreatedEvent() throws Interrupte
   }
 
   @Test
-  public void testRenameRequestProducesS3CreateAndDeleteEvents() throws InterruptedException, IOException {
+  public void testRenameRequestProducesRenameKeyEvent() throws InterruptedException, IOException {
     OmCompletedRequestInfo renameRequest = buildCompletedRequestInfo(4L, Type.RenameKey, "some/key4",
         new OperationArgs.RenameKeyArgs("some/key_RENAMED"));
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java

@@ -1344,7 +1344,7 @@ private List<OmVolumeArgs> listAllVolumes(String prefix, String startKey,
    * {@inheritDoc}
    */
   @Override
-  public List<OmCompletedRequestInfo> listCompletedRequestInfo(final String startKey,
+  public List<OmCompletedRequestInfo> listCompletedRequestInfo(final Long startKey,
                                                                final int maxResults)
       throws IOException {
     List<OmCompletedRequestInfo> results = new ArrayList<>();
@@ -1354,11 +1354,11 @@ public List<OmCompletedRequestInfo> listCompletedRequestInfo(final String startK
             tableIterator = getCompletedRequestInfoTable().iterator()) {
 
       boolean skipFirst = false;
-      if (StringUtils.isNotBlank(startKey)) {
+      if (startKey != null) {
         // TODO: what happens if the seek position is no longer
         // available?  Do we go to the end of the list
         // or the first key > startKey
-        tableIterator.seek(Long.valueOf(startKey));
+        tableIterator.seek(startKey);
         skipFirst = true;
       }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContextImpl.java

@@ -41,7 +41,7 @@ public boolean isLeaderReady() {
   // TODO: should we allow plugins to pass in maxResults or just limit
   // them to some predefined value for safety?  e.g. 10K
   @Override
-  public List<OmCompletedRequestInfo> listCompletedRequestInfo(String startKey, int maxResults) throws IOException {
+  public List<OmCompletedRequestInfo> listCompletedRequestInfo(Long startKey, int maxResults) throws IOException {
     return ozoneManager.getMetadataManager().listCompletedRequestInfo(startKey, maxResults);
   }