Bugfix for maxPendingProcessRecordsInput is 0

[lucienlu-aws]

Issue #, if available:

Description of changes:

• Add maxPendingProcessRecordsInput config to PollingConfig #1674 introduced a bug when users specify maxPendingProcessRecordsInput to be 0. This commit fixes it, details below
• Limit maxPendingProcessRecordsInput to have max value of 5
• [small fix] Update initial lastEventDeliveryTime to be now instead of epoch to avoid WARN log at KCL startup

                    "{}: Record delivery time to shard consumer is high at {} millis. Check the ExecutorStateEvent logs"
                            + " to see the state of the executor service. Also check if the RecordProcessor's processing "
                            + "time is high. ",

Symptom

When maxPendingProcessRecordsInput is 0 and the KCL application initializes a ShardConsumer at a checkpoint that is not at tip (i.e. could be TRIM_HORIZON with 2 batches of records or a sequence number that has at least two batches of records after it), there is a 60 second delay when retrieving the second record.

Fix

• When maxPendingProcessRecordsInput is 0, don't fetch records from Kinesis until Publisher has demand
• Do not attempt to drainQueue in subcsroption.request() and NotifyingSubscriber

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[chenylee-aws]

Could you update it to Instant.now()? This is to avoid the false large delay warning for the first batch of records

[lucienlu-aws]

Updated

[chenylee-aws]

should probably use DEFAULT_MAX_PENDING_PROCESS_RECORDS_INPUT_LIMIT here as well.

[lucienlu-aws]

Fixed

[chenylee-aws]

could you make these two private?

[lucienlu-aws]

Updated