feat: add support for OnSuccess sink (#3040)

Signed-off-by: Vaibhav Tiwari <[email protected]>
Signed-off-by: Vigith Maurice <[email protected]>
Co-authored-by: Vigith Maurice <[email protected]>

Author: vaibhavtiwari33

api/json-schema/schema.json

@@ -22485,6 +22485,10 @@
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.Log",
           "description": "Log sink is used to write the data to the log."
         },
+        "onSuccess": {
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.AbstractSink",
+          "description": "OnSuccess sink allows triggering a secondary sink operation only after the primary sink completes successfully The writes to OnSuccess sink will only be initiated if the ud-sink response field sets it. A new Message crafted in the Primary sink can be written on the OnSuccess sink."
+        },
         "pulsar": {
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.PulsarSink",
           "description": "Pulsar sink is used to write the data to the Apache Pulsar."

api/openapi-spec/swagger.json

@@ -22472,6 +22472,10 @@
           "description": "Log sink is used to write the data to the log.",
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.Log"
         },
+        "onSuccess": {
+          "description": "OnSuccess sink allows triggering a secondary sink operation only after the primary sink completes successfully The writes to OnSuccess sink will only be initiated if the ud-sink response field sets it. A new Message crafted in the Primary sink can be written on the OnSuccess sink.",
+          "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.AbstractSink"
+        },
         "pulsar": {
           "description": "Pulsar sink is used to write the data to the Apache Pulsar.",
           "$ref": "#/definitions/io.numaproj.numaflow.v1alpha1.PulsarSink"

config/base/crds/full/numaflow.numaproj.io_monovertices.yaml

@@ -12086,6 +12086,31 @@ it.
 
 <td>
 
+<code>onSuccess</code></br> <em>
+<a href="#numaflow.numaproj.io/v1alpha1.AbstractSink"> AbstractSink </a>
+</em>
+</td>
+
+<td>
+
+<em>(Optional)</em>
+<p>
+
+OnSuccess sink allows triggering a secondary sink operation only after
+the primary sink completes successfully The writes to OnSuccess sink
+will only be initiated if the ud-sink response field sets it. A new
+Message crafted in the Primary sink can be written on the OnSuccess
+sink.
+</p>
+
+</td>
+
+</tr>
+
+<tr>
+
+<td>
+
 <code>retryStrategy</code></br> <em>
 <a href="#numaflow.numaproj.io/v1alpha1.RetryStrategy"> RetryStrategy
 </a> </em>

config/base/crds/full/numaflow.numaproj.io_pipelines.yaml

@@ -63,6 +63,7 @@ const (
 	CtrUdf               = "udf"
 	CtrUdsink            = "udsink"
 	CtrFallbackUdsink    = "fb-udsink"
+	CtrOnSuccessUdsink   = "ons-udsink"
 	CtrUdsource          = "udsource"
 	CtrUdtransformer     = "transformer"
 	CtrUdSideInput       = "udsi"
@@ -73,13 +74,14 @@ const (
 	CtrUdStore           = "udstore"
 
 	// user-defined container types
-	UDContainerFunction     = "udf"
-	UDContainerSink         = "udsink"
-	UDContainerFallbackSink = "fb-udsink"
-	UDContainerTransformer  = "transformer"
-	UDContainerSource       = "udsource"
-	UDContainerSideInputs   = "udsi"
-	UDContainerStore        = "udstore"
+	UDContainerFunction      = "udf"
+	UDContainerSink          = "udsink"
+	UDContainerFallbackSink  = "fb-udsink"
+	UDContainerOnSuccessSink = "ons-udsink"
+	UDContainerTransformer   = "transformer"
+	UDContainerSource        = "udsource"
+	UDContainerSideInputs    = "udsi"
+	UDContainerStore         = "udstore"
 
 	// components
 	ComponentISBSvc            = "isbsvc"