Flaky-test: PulsarSinksTest.testKinesis / KinesisSinkTester.addMoreRecordsAndGetMillisBehindLatest

KinesisSinkTester.addMoreRecordsAndGetMillisBehindLatest is flaky. It fails sporadically.

[example failure](https://github.com/apache/pulsar/runs/6205142531?check_suite_focus=true#step:12:20926)

```
Error: testKinesis(org.apache.pulsar.tests.integration.io.sinks.PulsarSinksTest) Time elapsed: 62.576 s <<< FAILURE!
java.io.UncheckedIOException: Cannot encode string.
at software.amazon.awssdk.utils.StringUtils.fromBytes(StringUtils.java:578)
at software.amazon.awssdk.core.BytesWrapper.asString(BytesWrapper.java:82)
at org.apache.pulsar.tests.integration.io.sinks.KinesisSinkTester.addMoreRecordsAndGetMillisBehindLatest(KinesisSinkTester.java:210)
at org.apache.pulsar.tests.integration.io.sinks.KinesisSinkTester.lambda$internalValidateSinkResult$2(KinesisSinkTester.java:195)
at org.awaitility.core.CallableCondition$ConditionEvaluationWrapper.eval(CallableCondition.java:99)
at org.awaitility.core.ConditionAwaiter$ConditionPoller.call(ConditionAwaiter.java:248)
at org.awaitility.core.ConditionAwaiter$ConditionPoller.call(ConditionAwaiter.java:235)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: java.nio.charset.MalformedInputException: Input length = 3
at java.base/java.nio.charset.CoderResult.throwException(CoderResult.java:274)
at java.base/java.nio.charset.CharsetDecoder.decode(CharsetDecoder.java:820)
at software.amazon.awssdk.utils.StringUtils.fromBytes(StringUtils.java:576)
... 10 more
```

Start with KinesisSinkTester.java at addMoreRecordsAndGetMillisBehindLatest (line 210) and the PulsarSinksTest.testKinesis integration test. Reproduce the example failure and inspect the reported AWS SDK MalformedInputException during the Awaitility polling path. Done means the Kinesis sink test no longer fails sporadically and still validates the sink result.

Written by the indexing model from the issue text.