feat: add timeout for pausing pipeline. Fixes #992

api/json-schema/schema.json

@@ -18539,6 +18539,11 @@
         "desiredPhase": {
           "description": "DesiredPhase used to bring the pipeline from current phase to desired phase",
           "type": "string"
+        },
+        "pauseGracePeriodSeconds": {
+          "description": "PauseGracePeriodSeconds used to pause pipeline gracefully",
+          "format": "int32",
+          "type": "integer"
         }
       },
       "type": "object"

api/openapi-spec/swagger.json

@@ -18535,6 +18535,11 @@
         "desiredPhase": {
           "description": "DesiredPhase used to bring the pipeline from current phase to desired phase",
           "type": "string"
+        },
+        "pauseGracePeriodSeconds": {
+          "description": "PauseGracePeriodSeconds used to pause pipeline gracefully",
+          "type": "integer",
+          "format": "int32"
         }
       }
     },

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

@@ -97,6 +97,7 @@ spec:
                 default:
                   deleteGracePeriodSeconds: 30
                   desiredPhase: Running
+                  pauseGracePeriodSeconds: 30
                 properties:
                   deleteGracePeriodSeconds:
                     default: 30
@@ -113,6 +114,10 @@ spec:
                     - Paused
                     - Deleting
                     type: string
+                  pauseGracePeriodSeconds:
+                    default: 30
+                    format: int32
+                    type: integer
                 type: object
               limits:
                 default:

config/install.yaml

@@ -2626,6 +2626,7 @@ spec:
                 default:
                   deleteGracePeriodSeconds: 30
                   desiredPhase: Running
+                  pauseGracePeriodSeconds: 30
                 properties:
                   deleteGracePeriodSeconds:
                     default: 30
@@ -2642,6 +2643,10 @@ spec:
                     - Paused
                     - Deleting
                     type: string
+                  pauseGracePeriodSeconds:
+                    default: 30
+                    format: int32
+                    type: integer
                 type: object
               limits:
                 default:

config/namespace-install.yaml

@@ -2626,6 +2626,7 @@ spec:
                 default:
                   deleteGracePeriodSeconds: 30
                   desiredPhase: Running
+                  pauseGracePeriodSeconds: 30
                 properties:
                   deleteGracePeriodSeconds:
                     default: 30
@@ -2642,6 +2643,10 @@ spec:
                     - Paused
                     - Deleting
                     type: string
+                  pauseGracePeriodSeconds:
+                    default: 30
+                    format: int32
+                    type: integer
                 type: object
               limits:
                 default:

docs/APIs.md

@@ -2688,6 +2688,17 @@ phase
 </p>
 </td>
 </tr>
+<tr>
+<td>
+<code>pauseGracePeriodSeconds</code></br> <em> int32 </em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>
+PauseGracePeriodSeconds used to pause pipeline gracefully
+</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="numaflow.numaproj.io/v1alpha1.Log">

docs/user-guide/reference/pipeline-operations.md

@@ -22,7 +22,7 @@ To pause a pipeline, use the command below, it will bring the pipeline to `Pause
 
 Pausing a pipeline will not cause data loss. It does not clean up the unprocessed data in the pipeline, but just terminates the running pods. When the pipeline is resumed, the pods will be restarted and continue processing the unprocessed data.
 
-When pausing a pipeline, it will shutdown the source vertex pods first, and then wait for the other vertices to finish the backlog before terminating them.
+When pausing a pipeline, it will shutdown the source vertex pods first, and then wait for the other vertices to finish the backlog before terminating them. However, it will not wait forever and will terminate the pods after `pauseGracePeriodSeconds`. This is default set to 30 and can be customized by setting `spec.lifecycle.pauseGracePeriodSeconds`.
 
 If there's a [reduce](../user-defined-functions/reduce/reduce.md) vertex in the pipeline, please make sure it uses [Persistent Volume Claim](../user-defined-functions/reduce/reduce.md#persistent-volume-claim-pvc) for storage, otherwise the data will be lost.
 

pkg/apis/numaflow/v1alpha1/const.go

@@ -39,6 +39,7 @@ const (
 	KeyVertexName       = "numaflow.numaproj.io/vertex-name"
 	KeyReplica          = "numaflow.numaproj.io/replica"
 	KeySideInputName    = "numaflow.numaproj.io/side-input-name"
+	KeyPauseTimestamp   = "numaflow.numaproj.io/pause-timestamp"
 	KeyDefaultContainer = "kubectl.kubernetes.io/default-container"
 
 	// ID key in the header of sources like http