kafka consumer child pipeline can not be same as parent pipeline. apa…

…che#2578 apache#4326
bamaer · Sep 19, 2024 · 13e24fa · 13e24fa
1 parent 59eadc6
commit 13e24fa
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 0 deletions.
diff --git a/...main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java b/...main/java/org/apache/hop/pipeline/transforms/kafka/consumer/KafkaConsumerInputDialog.java
@@ -31,6 +31,7 @@
 import org.apache.hop.core.exception.HopException;
 import org.apache.hop.core.gui.Point;
 import org.apache.hop.core.row.value.ValueMetaFactory;
+import org.apache.hop.core.util.Utils;
 import org.apache.hop.core.variables.IVariables;
 import org.apache.hop.i18n.BaseMessages;
 import org.apache.hop.pipeline.PipelineMeta;
@@ -43,6 +44,7 @@
 import org.apache.hop.ui.core.PropsUi;
 import org.apache.hop.ui.core.dialog.BaseDialog;
 import org.apache.hop.ui.core.dialog.ErrorDialog;
+import org.apache.hop.ui.core.dialog.MessageBox;
 import org.apache.hop.ui.core.gui.GuiResource;
 import org.apache.hop.ui.core.gui.WindowProperty;
 import org.apache.hop.ui.core.widget.ColumnInfo;
@@ -268,6 +270,22 @@ public void widgetSelected(SelectionEvent e) {
 
   private void ok() {
     transformName = wTransformName.getText();
+    if (Utils.isEmpty(wFilename.getText())) {
+      MessageBox mb = new MessageBox(shell, SWT.OK | SWT.ICON_ERROR);
+      mb.setText(BaseMessages.getString(PKG, "KafkaConsumerInputDialog.FilenameMissing.Header"));
+      mb.setMessage(
+          BaseMessages.getString(PKG, "KafkaConsumerInputDialog.FilenameMissing.Message"));
+      mb.open();
+      return;
+    }
+    if (isSelfReferencing()) {
+      MessageBox mb = new MessageBox(shell, SWT.OK | SWT.ICON_ERROR);
+      mb.setText(BaseMessages.getString(PKG, "KafkaConsumerInputDialog.SelfReference.Header"));
+      mb.setMessage(BaseMessages.getString(PKG, "KafkaConsumerInputDialog.SelfReference.Message"));
+      mb.open();
+      return;
+    }
+
     updateMeta(meta);
     dispose();
   }
@@ -948,4 +966,13 @@ protected void populateSubTransforms(Event event) {
       log.logError("Error getting transform names from Kafka pipeline", e);
     }
   }
+
+  private boolean isSelfReferencing() {
+    if (variables
+        .resolve(wFilename.getText())
+        .equals(variables.resolve(pipelineMeta.getFilename()))) {
+      return true;
+    }
+    return false;
+  }
 }
diff --git a/...rces/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties b/...rces/org/apache/hop/pipeline/transforms/kafka/consumer/messages/messages_en_US.properties
@@ -72,3 +72,7 @@ KafkaConsumerInputMeta.Injection.TRANSFORMATION_PATH=The path to the sub-pipelin
 KafkaConsumerInputMeta.Injection.VALUES=The value(s) of the Kafka consumer configuration properties.
 KafkaConsumerInputMeta.keyword=kafka,consumer,input
 KafkaConsumerInputMeta.UnableToCreateValueType=Unable to create output field value for {0}
+KafkaConsumerInputDialog.FilenameMissing.Header=Warning
+KafkaConsumerInputDialog.FilenameMissing.Message=The Kafka pipeline filename is missing.
+KafkaConsumerInputDialog.SelfReference.Header=Warning
+KafkaConsumerInputDialog.SelfReference.Message=This pipeline can''t be used as the Kafka pipeline. Please choose another Kafka pipeline.