[chenylee-aws]

Description of changes:

Previously, if anything in KCL's data path threw before a batch was handed to processRecords (shard lookup, deaggregation, schema decoding, task execution listener, metrics, etc.), ShardConsumerSubscriber.onNext logged "The incoming data has been skipped" and requested the next batch anyway. The batch was never delivered, and once the application checkpointed a later batch those records were lost for good. Failures inside ProcessTask were also swallowed and reported as a completed task, and at shard end the exception was discarded entirely.

With this change, the handoff to the record processor is the only thing that advances the read position. Every failure in KCL's own handling of a batch now leads to either redelivery or a halted shard consumer with the error surfaced:

• Exceptions before the handoff: ProcessTask rethrows instead of swallowing, and the subscriber retries the same batch in place with exponential backoff (500 ms to 10 s). Ordering is preserved, because the next batch is not requested until the current one is handed off. lastAccepted and request(1) only advance after a successful handoff.
• Errors (OOM, linkage errors, etc.): not retried. The subscriber stops without advancing, and the error is recorded in dispatchFailure, which healthCheck rethrows on the Scheduler thread, as before.
• Shutdown or interrupt during retry: retrying stops. The batch was never delivered and is reprocessed from the last checkpoint by the next lease owner.
• Exceptions from the application's processRecords: unchanged. They are logged and the batch is skipped.

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