feat: Generate Idle Watermark if the source is idling

[chandankumar4]

Explain what this PR does.

• Generate the idle Watermark if the source is idling

Fixes #633

Example Pipeline to test the changes:

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: idle-source
spec:
  watermark:
    idleSource:
      threshold: 5s
      incrementBy: 5s
      stepInterval: 5s
  limits:
    readBatchSize: 50
  vertices:
    - name: in
      scale:
        min: 2
      source:
        http: {}
    - name: atoi
      scale:
        min: 1
      udf:
        container:
          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd
          image: quay.io/numaio/numaflow-go/map-even-odd:v0.5.0
    - name: compute-sum
      partitions: 2
      udf:
        container:
          # Compute the sum, see https://github.com/numaproj/numaflow-go/tree/main/pkg/reducer/examples/sum
          image: quay.io/numaio/numaflow-go/reduce-sum:v0.5.0
        groupBy:
          window:
            fixed:
              length: 10s
          keyed: true
          storage:
            persistentVolumeClaim:
              volumeSize: 10Gi
              accessMode: ReadWriteOnce
    - name: sink
      scale:
        min: 1
      sink:
        udsink:
          container:
            # A redis sink for e2e testing, see https://github.com/numaproj/numaflow-sinks/tree/main/redis-e2e-test-sink
            image: quay.io/numaio/numaflow-sink/redis-e2e-test-sink:v0.5.0
  edges:
    - from: in
      to: atoi
    - from: atoi
      to: compute-sum
    - from: compute-sum
      to: sink

Note:
This will not work in below case:

• In case of Kafka with 12 partition and 6 partition is assigned to one pod and rest 6 to another, now If only 3 of them is receiving the data out of 6 partition.

[KeranYang]

Can we add some comments on these numbers on how they are used to test source idling? The comment should make it clear why removing these configurations will break the test.

[KeranYang]

I am un-resolving this one. We configured the idle source, updated the simple keyed reduce test with a new start time and new wait time before sending next batch.

I have a hard time understanding how this test works in terms of verifying idle source works. Can we write some comments on why this test is testing the source idling functionality? How these numbers 5s, 3s, 2s, startTime := 100, time.Sleep(10 * time.Millisecond)andWithTimeout(120*time.Second)` work together to achieve idle source test?

[vigith]

#1409

[vigith]

can we add a unit test of idle handler?

[vigith]

please make sure this doesn't involve a n/w call. i am quite sure it wouldn't.

[vigith]

this is not right, we only append by never remove. we will have to ask users to add Partitions() to ud-source interface.

@KeranYang WDYT?

[vigith]

#1405

[KeranYang]

I agree this is not right. With this change we now have two sources of truth on how many partitions a source has. One is the read batch and the other is the source Partitions() interface. Should stick with the second one.

The removing part can be tricky because if we remove a certain partition, we also need to remove the corresponding watermark KV store etc. We can look into details as part of issue 1405 that you created.

[vigith]

please document else branch

[vigith]

need few changes as discussed.

[KeranYang]

I have only one minor comment open. Approving.