fix(kafka-setup): skip schema registry config setup when kafka setup is disabled (#15268)

Author: chakru-r

datahub-upgrade/src/main/java/com/linkedin/datahub/upgrade/system/kafka/ConfluentSchemaRegistryCleanupPolicyStep.java

@@ -47,9 +47,16 @@ public Function<UpgradeContext, UpgradeStepResult> executable() {
     return (context) -> {
       log.info("Configuring Confluent Schema Registry cleanup policies...");
 
-      // Check if Confluent Schema Registry is enabled
+      // Check if Kafka setup is enabled
       if (_kafkaConfiguration.getSetup() == null
-          || !_kafkaConfiguration.getSetup().isUseConfluentSchemaRegistry()) {
+          || !_kafkaConfiguration.getSetup().isPreCreateTopics()) {
+        log.info(
+            "Skipping Confluent Schema Registry cleanup policy configuration - Kafka setup is disabled");
+        return new DefaultUpgradeStepResult(this.id(), DataHubUpgradeState.SUCCEEDED);
+      }
+
+      // Check if Confluent Schema Registry is enabled
+      if (!_kafkaConfiguration.getSetup().isUseConfluentSchemaRegistry()) {
         log.info(
             "Skipping Confluent Schema Registry cleanup policy configuration - schema registry is disabled");
         return new DefaultUpgradeStepResult(this.id(), DataHubUpgradeState.SUCCEEDED);

datahub-upgrade/src/test/java/com/linkedin/datahub/upgrade/system/kafka/ConfluentSchemaRegistryCleanupPolicyStepTest.java

@@ -82,6 +82,18 @@ public void testSkipWhenSetupConfigIsNull() {
     assertEquals(result.result(), DataHubUpgradeState.SUCCEEDED);
   }
 
+  @Test
+  public void testSkipWhenPreCreateTopicsDisabled() {
+    // Set preCreateTopics to false
+    kafkaConfiguration.getSetup().setPreCreateTopics(false);
+
+    UpgradeContext mockContext = mock(UpgradeContext.class);
+    UpgradeStepResult result = step.executable().apply(mockContext);
+
+    assertNotNull(result);
+    assertEquals(result.result(), DataHubUpgradeState.SUCCEEDED);
+  }
+
   @Test
   public void testSuccessfulCleanupPolicyConfiguration() throws Exception {
     // Create spy to mock the createAdminClient method