fix: initialize inflightAcks channel to not nil channel

[afugazzotto]

Closes #1529

Initialize inflightAcks channel to not nil channel to avoid hanging on cleanup function when there are no messages to acknowledge.

How to Test/Reproduce:

1. setup a Kafka client and a Kafka topic with 10 partitions
2. in a bash script, set Kafka client to send 10 messages per second
3. setup a Numaflow pipeline with source vertex to consume from the Kafka client (disable autoscaling)
4. manually scale source vertex to 12 or more pods (kubectl scale vtx <vtx_name> --replicas=12)
5. manually delete pod-0
6. wait for a new pod-0 to restart
7. verify it has no topics assigned to it in the metrics: kafka_source_read_total
8. scale down to 1 pod
9. at this point, before the fix, the messages flow to the log sink would stop. After the fix, pod-0 is not stuck in the cleanup state and can resume consuming messages (kafka_source_read_total should show for all 10 partitions)

[KeranYang]

Why do we need inflightAcks at first place? My understanding is Ack is a blocking call hence inflightAcks should be either nil or closed. Am I missing anything? cc: @vigith

[afugazzotto]

Why do we need inflightAcks at first place? My understanding is Ack is a blocking call hence inflightAcks should be either nil or closed. Am I missing anything? cc: @vigith

Based on this question, would it make more sense (or be preferred) to remove the inflight ack "check" in the Cleanup function (https://github.com/numaproj/numaflow/blob/main/pkg/sources/kafka/handler.go#L59)?

[KeranYang]

would it make more sense (or be preferred) to remove the inflight ack "check" in the Cleanup function

or completely remove the var inflightAcks?

[vigith]

Why do we need inflightAcks at first place? My understanding is Ack is a blocking call hence inflightAcks should be either nil or closed. Am I missing anything? cc: @vigith

Cleanup function is not called "synchronously" with the Ack function, so there will be a state where Ack is running and Cleanup has been called.