[SPARK-53904][SS] Use Consumer.poll(Duration) instead of Consumer.poll(long)

[dongjoon-hyun]

What changes were proposed in this pull request?

This PR aims to use Consumer.poll(Duration) instead of Consumer.poll(long).

• https://kafka.apache.org/documentation.html#upgrade_400_notable_consumer

  The poll(long) method was removed from the consumer. Please use poll(Duration) instead. Note that there is a difference in behavior between the two methods. The poll(Duration) method does not block beyond the timeout awaiting partition assignment, whereas the earlier poll(long) method used to wait beyond the timeout.

This PR is consistent with KAFKA-15908's approach which uses Duration.ZERO and Duration.ofMillis(500) together. In general, Duration.ZERO is used like the following except KafkaOffsetReaderConsumer (and DirectKafkaStreamSuite) classes which use Duration.ofMillis(500) to make CI passes.

- consumer.poll(0)
+ consumer.poll(Duration.ZERO)

Why are the changes needed?

• Apache Kafka 2.0.0 deprecated Consumer.poll(long) via KIP-266: Fix consumer indefinite blocking behavior.

  The pre-existing variant poll(long timeout) would block indefinitely for metadata updates if they were needed, then it would issue a fetch and poll for timeout ms for new records. The initial indefinite metadata block caused applications to become stuck when the brokers became unavailable. The existence of the timeout parameter made the indefinite block especially unintuitive.

• Apache Kafka 4.0.0 removed it at KAFKA-15908.

  • KAFKA-15908: Remove deprecated Consumer.poll(long timeout) kafka#17368

Does this PR introduce any user-facing change?

No behavior change.

How was this patch tested?

Pass the CIs.

Was this patch authored or co-authored using generative AI tooling?

No.