current getPartition impl causing watermark issues for partitioned topics

[ccm32004]

When testing a consumer pipeline with a reduce vertex, consuming from a topic with multiple partitions, it was found that the reduce vertex would fail to close its defined window, and no messages would be sent downstream to the out vertex.

Testing notes:
The test image was built from the latest commit on main: d014fdd ("chore: update docs (#85)")
apache pulsar image uses numaflow-java v0.11.0
(also tested with numaflow-java v.0.12.0, messages were still not sent downstream)

configured a streamnative apache pulsar topic with 5 partitions
configured a producer pipeline to publish message to this topic, with low tps (1 rpu over duration of 5 seconds) and high tps (10000 rpu over duration of 1 seconds)

the producer pipelines specs (rpu, duration was changed depending on high or low tps)

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: producer-pipeline
spec:
  vertices:
    - name: in
      source:
        generator:
          rpu: 1
          duration: 5s
          valueBlob: "anVzdCBhIHN0cmluZw=="
    - name: out
      scale:
        min: 1
        max: 1
      volumes:
        - name: pulsar-config-volume
          configMap:
            name: producer-config
            items:
              - key: application.yml
                path: application.yml
      sink:
        udsink:
          container:
            image: apache-pulsar-java:v0.3.0
            args: [ "--config=file:/conf/application.yml" ]
            imagePullPolicy: Never
            envFrom:
              - secretRef:
                  name: pulsar-auth-secret
            volumeMounts:
              - name: pulsar-config-volume
                mountPath: /conf
  edges:
    - from: in
      to: out

producer config

apiVersion: v1
kind: ConfigMap
metadata:
  name: producer-config
data:
  application.yml: |
    pulsar:
      client:
        clientConfig:
          serviceUrl: "https://#####.aws-use2-production-snci-pool-kid.streamnative.aws.snio.cloud"
          authPluginClassName: org.apache.pulsar.client.impl.auth.AuthenticationToken
          authParams: "${PULSAR_AUTH_TOKEN}"
      producer:
        enabled: true
        useAutoProduceSchema: false
        dropInvalidMessages: false
        producerConfig:
          topicName: "persistent://demo_cece/dev/test-topic"
          batchingMaxPublishDelayMicros: 100000
          batchingMaxMessages: 5000
      admin:
        adminConfig:
          serviceUrl: "https://#####.aws-use2-production-snci-pool-kid.streamnative.aws.snio.cloud"
          authPluginClassName: org.apache.pulsar.client.impl.auth.AuthenticationToken
          authParams: "${PULSAR_AUTH_TOKEN}"

the consumer pipeline specs (idle configuration enabled):

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: consumer-pipeline-idle-enabled
spec:
  vertices:
    - name: in
      source:
        udsource:
          container:
            image: apache-pulsar-java:before-partition
            args: [ "--config=file:/conf/application.yml" ]
            imagePullPolicy: Never
            resources:
              requests:
                cpu: 1500m
                memory: 640Mi
              limits:
                cpu: 1500m
                memory: 640Mi
            envFrom:
              - secretRef:
                  name: pulsar-auth-secret
            volumeMounts:
              - name: pulsar-config-volume
                mountPath: /conf
      limits:
        readBatchSize: 1
      volumes:
        - name: pulsar-config-volume
          configMap:
            name: consumer-config
            items:
              - key: application.yml
                path: application.yml
    - name: reduce
      udf:
        container:
          image: quay.io/numaio/numaflow-go/reduce-counter:stable
          imagePullPolicy: Always
        groupBy:
          window:
            fixed:
              length: 30s
          keyed: false
          storage:
            persistentVolumeClaim:
              volumeSize: 1Gi
    - name: out
      scale:
        max: 3
      sink:
        log: {}
  watermark:
    maxDelay: 5s
    idleSource:
      threshold: 15s
      incrementBy: 3s
      stepInterval: 2s
  edges:
    - from: in
      to: reduce
    - from: reduce
      to: out

consumer config

apiVersion: v1
kind: ConfigMap
metadata:
  name: consumer-config
data:
  application.yml: |
    pulsar:
      client:
        clientConfig:
          serviceUrl: "https://#####.aws-use2-production-snci-pool-kid.streamnative.aws.snio.cloud"
          authPluginClassName: org.apache.pulsar.client.impl.auth.AuthenticationToken
          authParams: "${PULSAR_AUTH_TOKEN}"
      consumer:
        enabled: true
        useAutoConsumeSchema: true
        consumerConfig:
          topicNames: "persistent://demo_cece/dev/test-topic"
          subscriptionName: "my-consumer-sub"
          receiverQueueSize: 500
          subscriptionInitialPosition: Earliest
      admin:
        adminConfig:
          serviceUrl: "https://#####.aws-use2-production-snci-pool-kid.streamnative.aws.snio.cloud"
          authPluginClassName: org.apache.pulsar.client.impl.auth.AuthenticationToken
          authParams: "${PULSAR_AUTH_TOKEN}"

tested with numaflow v1.7.0 and numaflow v1.7.4 (latest released numaflow version as of writing)
low tps with numaflow v1.7.0
reduce vertex numa logs

in one of the runs, looked into the watermark processor creation, found these logs, five processors were created for the five partitions

out vertex numa logs

high tps with numaflow v1.7.0
reduce vertex numa logs

out vertex numa logs

low tps with numaflow v1.7.4
As can be seen, watermarks are not updating properly, four out of the five watermarks for the partitions are not working, at -1

high tps with numaflow v1.7.4