Add an admin job for rewriting Galapagos internal topics

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/AbstractAdminJob.java

@@ -0,0 +1,46 @@
+package com.hermesworld.ais.galapagos.adminjobs.impl;
+
+import com.hermesworld.ais.galapagos.adminjobs.AdminJob;
+
+public abstract class AbstractAdminJob implements AdminJob {
+
+    private final static int BANNER_WIDTH = 80;
+
+    private final static int MIN_BANNER_PREFIX = 3;
+
+    protected String banner(String text) {
+        return banner(text, BANNER_WIDTH);
+    }
+
+    protected String banner(String text, int bannerWidth) {
+        if (text.length() > (bannerWidth - (MIN_BANNER_PREFIX + 1) * 2)) {
+            return "=== " + text + " ===";
+        }
+        int preLen = (bannerWidth - text.length()) / 2;
+        int postLen = bannerWidth - text.length() - preLen;
+
+        StringBuilder sb = new StringBuilder();
+        if (text.isEmpty()) {
+            return "=".repeat(bannerWidth);
+        }
+        sb.append("=".repeat(preLen - 1));
+        sb.append(" ");
+        sb.append(text);
+        sb.append(" ");
+        sb.append("=".repeat(postLen - 1));
+        return sb.toString();
+    }
+
+    protected void printBanner(String text) {
+        System.out.println();
+        System.out.println(banner(text));
+        System.out.println();
+    }
+
+    protected void printBanner(String text, int bannerWidth) {
+        System.out.println();
+        System.out.println(banner(text, bannerWidth));
+        System.out.println();
+    }
+
+}

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/CleanupDeveloperAuthenticationsJob.java

@@ -1,12 +1,11 @@
 package com.hermesworld.ais.galapagos.adminjobs.impl;
 
-import com.hermesworld.ais.galapagos.adminjobs.AdminJob;
 import com.hermesworld.ais.galapagos.devauth.DeveloperAuthenticationService;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.stereotype.Component;
 
 @Component
-public class CleanupDeveloperAuthenticationsJob implements AdminJob {
+public class CleanupDeveloperAuthenticationsJob extends AbstractAdminJob {
 
     private final DeveloperAuthenticationService developerAuthenticationService;
 
@@ -21,14 +20,11 @@ public String getJobName() {
 
     @Override
     public void run(ApplicationArguments allArguments) throws Exception {
-        System.out.println();
-        System.out.println(
-                "=========== Starting Cleanup of expired Developer Authentications on all Kafka clusters ===========");
-        System.out.println();
+        printBanner("Starting Cleanup of expired Developer Authentications on all Kafka clusters");
 
-        System.out.println("=========== Cleanup of total "
+        printBanner("Cleanup of total "
                 + developerAuthenticationService.clearExpiredDeveloperAuthenticationsOnAllClusters().get()
-                + " expired Developer Certificates on all Kafka clusters was successful ===========");
+                + " expired Developer Certificates on all Kafka clusters was successful");
     }
 
 }

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/CreateBackupJob.java

@@ -2,7 +2,6 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.hermesworld.ais.galapagos.adminjobs.AdminJob;
 import com.hermesworld.ais.galapagos.kafka.KafkaCluster;
 import com.hermesworld.ais.galapagos.kafka.KafkaClusters;
 import com.hermesworld.ais.galapagos.kafka.util.TopicBasedRepository;
@@ -18,7 +17,7 @@
 import java.util.Optional;
 
 @Component
-public class CreateBackupJob implements AdminJob {
+public class CreateBackupJob extends AbstractAdminJob {
 
     private final KafkaClusters kafkaClusters;
 
@@ -43,16 +42,12 @@ public void run(ApplicationArguments allArguments) throws Exception {
 
         JSONObject backup = new JSONObject();
 
-        System.out.println();
-        System.out.println("========================= Starting Backup Creation ========================");
-        System.out.println();
+        printBanner("Starting Backup Creation");
 
         kafkaClusters.getEnvironmentIds().forEach(envId -> kafkaClusters.getEnvironment(envId)
                 .ifPresent(env -> backup.put(envId, backupEnvironment(env))));
 
-        System.out.println();
-        System.out.println("========================= Backup Creation COMPLETE ========================");
-        System.out.println();
+        printBanner("Backup Creation COMPLETE");
 
         if (!createBackupFile) {
             System.out.println("Backup JSON:");
@@ -69,6 +64,7 @@ public void run(ApplicationArguments allArguments) throws Exception {
             }
             catch (IOException e) {
                 System.err.println("Could not create Backup file");
+                // noinspection CallToPrintStackTrace
                 e.printStackTrace();
                 return;
             }
@@ -99,6 +95,7 @@ private JSONObject backupTopicData(TopicBasedRepository<? extends HasKey> repo)
                 result.put(obj.key(), new JSONObject(objectMapper.writeValueAsString(obj)));
             }
             catch (JSONException | JsonProcessingException e) {
+                // noinspection CallToPrintStackTrace
                 e.printStackTrace();
             }
         }

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/DeleteAclsJob.java

@@ -1,9 +1,5 @@
 package com.hermesworld.ais.galapagos.adminjobs.impl;
 
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Optional;
-
 import com.hermesworld.ais.galapagos.kafka.KafkaCluster;
 import com.hermesworld.ais.galapagos.kafka.KafkaClusters;
 import com.hermesworld.ais.galapagos.kafka.KafkaUser;
@@ -12,6 +8,10 @@
 import org.springframework.stereotype.Component;
 import org.springframework.util.ObjectUtils;
 
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Optional;
+
 /**
  * Admin job to explicitly delete ACLs from a Kafka Cluster. This job is useful if something went terribly wrong with
  * Galapagos, or if some rights have to be revoked quickly. <br>
@@ -50,13 +50,10 @@ public void runOnCluster(KafkaCluster cluster, ApplicationArguments allArguments
 
         cluster.removeUserAcls(new DummyKafkaUser(certificateDn)).get();
 
-        System.out.println();
-        System.out.println("========================== Certificate ACLs DELETED ==========================");
-        System.out.println();
+        printBanner("Certificate ACLs DELETED");
         System.out.println("All ACLs for certificate " + certificateDn + " have been deleted on Kafka Environment "
                 + cluster.getId());
-        System.out.println();
-        System.out.println("==============================================================================");
+        printBanner("");
     }
 
     private static class DummyKafkaUser implements KafkaUser {

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/GenerateToolingApiKeyJob.java

@@ -88,9 +88,7 @@ public void runOnCluster(KafkaCluster cluster, ApplicationArguments allArguments
         System.out.println("SAML Username: " + samlUsername);
         System.out.println("Secret: " + secret);
 
-        System.out.println();
-        System.out.println("==================== Galapagos Tooling API Key CREATED ====================");
-        System.out.println();
+        printBanner("Galapagos Tooling API Key CREATED");
 
         System.out.println("You can now use the API Key above for Galapagos external tooling on " + metadata.getName());
 
@@ -108,7 +106,7 @@ public void runOnCluster(KafkaCluster cluster, ApplicationArguments allArguments
         System.out.println(
                 "To remove ACLs for this API Key AND to delete the key itself, run Galapagos admin task galapagos.jobs.delete-apikey");
         System.out.println("with --kafka.environment=" + cluster.getId());
-        System.out.println();
-        System.out.println("==============================================================================");
+
+        printBanner("");
     }
 }

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/GenerateToolingCertificateJob.java

@@ -130,9 +130,7 @@ public void runOnCluster(KafkaCluster cluster, ApplicationArguments allArguments
             System.out.println("CERTIFICATE DATA: " + base64Data);
         }
 
-        System.out.println();
-        System.out.println("==================== Galapagos Tooling Certificate CREATED ====================");
-        System.out.println();
+        printBanner("Galapagos Tooling Certificate CREATED");
         if (!ObjectUtils.isEmpty(outputFilename)) {
             System.out.println("You can now use the certificate in " + outputFilename
                     + " for Galapagos external tooling on " + metadata.getName());
@@ -157,8 +155,8 @@ public void runOnCluster(KafkaCluster cluster, ApplicationArguments allArguments
         System.out.println("To remove ACLs for this certificate, run Galapagos admin task galapagos.jobs.delete-acls");
         System.out.println("with --certificate.dn=" + result.getPublicAuthenticationData().getString("dn")
                 + " --kafka.environment=" + cluster.getId());
-        System.out.println();
-        System.out.println("==============================================================================");
+
+        printBanner("");
     }
 
 }

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/ImportBackupJob.java

@@ -1,7 +1,6 @@
 package com.hermesworld.ais.galapagos.adminjobs.impl;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.hermesworld.ais.galapagos.adminjobs.AdminJob;
 import com.hermesworld.ais.galapagos.kafka.KafkaCluster;
 import com.hermesworld.ais.galapagos.kafka.KafkaClusters;
 import com.hermesworld.ais.galapagos.kafka.util.TopicBasedRepository;
@@ -37,7 +36,7 @@
  */
 
 @Component
-public class ImportBackupJob implements AdminJob {
+public class ImportBackupJob extends AbstractAdminJob {
 
     private final KafkaClusters kafkaClusters;
 
@@ -76,9 +75,7 @@ public void run(ApplicationArguments allArguments) throws Exception {
             data = new JSONObject(StreamUtils.copyToString(fis, StandardCharsets.UTF_8));
         }
 
-        System.out.println();
-        System.out.println("========================= Starting Backup Import ========================");
-        System.out.println();
+        printBanner("Starting Backup Import");
 
         Iterator<String> envIds = data.keys();
 
@@ -100,9 +97,7 @@ public void run(ApplicationArguments allArguments) throws Exception {
             importBackup(env, data.getJSONObject(envId));
         }
 
-        System.out.println();
-        System.out.println("========================= Backup Import COMPLETE ========================");
-        System.out.println();
+        printBanner("Backup Import COMPLETE");
     }
 
     @SuppressWarnings({ "unchecked", "rawtypes" })

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/ImportKnownApplicationsJob.java

@@ -1,18 +1,8 @@
 package com.hermesworld.ais.galapagos.adminjobs.impl;
 
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.util.*;
-import java.util.stream.Collectors;
-
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.type.TypeFactory;
-import com.hermesworld.ais.galapagos.adminjobs.AdminJob;
 import com.hermesworld.ais.galapagos.applications.BusinessCapability;
 import com.hermesworld.ais.galapagos.applications.impl.KnownApplicationImpl;
 import com.hermesworld.ais.galapagos.kafka.KafkaClusters;
@@ -24,6 +14,15 @@
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StreamUtils;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.*;
+import java.util.stream.Collectors;
+
 /**
  * Admin job to import known applications from a JSON file (or STDIN) to the global Galapagos topic
  * <code>known-applications</code>. This job can be used for development or test instances of Galapagos to import test
@@ -41,9 +40,9 @@
  */
 @Component
 @Slf4j
-public class ImportKnownApplicationsJob implements AdminJob {
+public class ImportKnownApplicationsJob extends AbstractAdminJob {
 
-    private KafkaClusters kafkaClusters;
+    private final KafkaClusters kafkaClusters;
 
     public ImportKnownApplicationsJob(KafkaClusters kafkaClusters) {
         this.kafkaClusters = kafkaClusters;
@@ -57,11 +56,10 @@ public String getJobName() {
     @Override
     public void run(ApplicationArguments allArguments) throws Exception {
         String jsonFile = Optional.ofNullable(allArguments.getOptionValues("applications.import.file"))
-                .map(ls -> ls.stream().findFirst().orElse(null)).orElse(null);
+                .flatMap(ls -> ls.stream().findFirst()).orElse(null);
 
         boolean remove = Optional.ofNullable(allArguments.getOptionValues("remove.missing.applications"))
-                .map(ls -> ls.stream().findFirst().orElse(null)).map(s -> s == null ? false : Boolean.parseBoolean(s))
-                .orElse(false);
+                .flatMap(ls -> ls.stream().findFirst()).map(s -> Boolean.parseBoolean(s)).orElse(false);
 
         if (ObjectUtils.isEmpty(jsonFile)) {
             throw new IllegalArgumentException("Please provide --applications.import.file=<file> for JSON to import");
@@ -111,16 +109,13 @@ public void run(ApplicationArguments allArguments) throws Exception {
             }
         }
 
-        System.out.println();
-        System.out.println("========================= Known applications IMPORTED ========================");
-        System.out.println();
+        printBanner("Known applications IMPORTED");
         System.out.println(cntImported + " new application(s) imported.");
         if (remove) {
             System.out.println(cntDeleted + " application(s) removed as they did not exist in JSON data.");
         }
-        System.out.println();
-        System.out.println("==============================================================================");
 
+        printBanner("");
     }
 
     private List<KnownApplicationImpl> readFromStdin() throws IOException {
@@ -170,7 +165,6 @@ private boolean businessCapabilityIsEqual(List<BusinessCapability> imported, Lis
         }
 
         return true;
-
     }
 
 }

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/MarkTopicApprovalRequiredJob.java

@@ -1,18 +1,17 @@
 package com.hermesworld.ais.galapagos.adminjobs.impl;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-
-import com.hermesworld.ais.galapagos.adminjobs.AdminJob;
 import com.hermesworld.ais.galapagos.kafka.KafkaClusters;
 import com.hermesworld.ais.galapagos.topics.service.TopicService;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.stereotype.Component;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
 @Component
-public class MarkTopicApprovalRequiredJob implements AdminJob {
+public class MarkTopicApprovalRequiredJob extends AbstractAdminJob {
 
     private final KafkaClusters kafkaClusters;
 
@@ -56,11 +55,8 @@ public void run(ApplicationArguments allArguments) throws Exception {
             throw new IllegalStateException("Could not find any of the specified topics on any environment");
         }
 
-        System.out.println();
-        System.out.println("============================ Topic(s) reconfigured ===========================");
-        System.out.println();
+        printBanner("Topic(s) reconfigured");
         resultLines.forEach(System.out::println);
-        System.out.println();
-        System.out.println("==============================================================================");
+        printBanner("");
     }
 }

src/main/java/com/hermesworld/ais/galapagos/adminjobs/impl/ResetApplicationPrefixesJob.java

@@ -61,12 +61,12 @@ protected void runOnCluster(KafkaCluster cluster, ApplicationArguments allArgume
                 .orElseThrow(() -> new IllegalArgumentException("Please provide required parameter --application.id"));
 
         try {
-            System.out.println("===== Resetting Prefixes and ACLs for Application " + applicationId + " =====");
+            printBanner("Resetting Prefixes and ACLs for Application " + applicationId);
             applicationsService.resetApplicationPrefixes(cluster.getId(), applicationId)
                     .thenCompose(metadata -> cluster.updateUserAcls(new ToolingUser(metadata, cluster.getId(),
                             kafkaClusters.getAuthenticationModule(cluster.getId()).orElseThrow(), aclSupport)))
                     .get();
-            System.out.println("===== Prefixes and ACL Reset SUCCESSFUL =====");
+            printBanner("Prefixes and ACL Reset SUCCESSFUL");
         }
         catch (ExecutionException e) {
             if (e.getCause() instanceof Exception) {