Kinesis ApiCallTimeout during the ingestion

[itsmesrds]

i'm using pinot 0.8.0 master branch.

Currently my producer is sending the records for 10 seconds and in the mean time the producer is idle and its throwing out the below error. same error is being thrown continuously for every 30sec, event after producer stopped.

2021/08/01 18:01:30.705 INFO [LLRealtimeSegmentDataManager_realtime_events__9__0__20210801T1750Z] [realtime_events__9__0__20210801T1750Z] Consumed 0 events from (rate:0.0/s), currentOffset={"shardId-000000000009":"49619229921359620555077470674058986369313634779052114066"}, numRowsConsumedSoFar=0, numRowsIndexedSoFar=0
2021/08/01 18:01:30.705 WARN [KinesisConsumer] [pool-22-thread-1] Task aborted due to exception.
shaded.software.amazon.awssdk.core.exception.AbortedException: Thread was interrupted
	at shaded.software.amazon.awssdk.core.exception.AbortedException$BuilderImpl.build(AbortedException.java:84) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.exception.AbortedException.create(AbortedException.java:38) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.handleInterruptedException(ApiCallTimeoutTrackingStage.java:143) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.translatePipelineException(ApiCallTimeoutTrackingStage.java:107) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.execute(ApiCallTimeoutTrackingStage.java:62) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.execute(ApiCallTimeoutTrackingStage.java:42) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallMetricCollectionStage.execute(ApiCallMetricCollectionStage.java:48) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallMetricCollectionStage.execute(ApiCallMetricCollectionStage.java:31) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.stages.ExecutionFailureExceptionReportingStage.execute(ExecutionFailureExceptionReportingStage.java:37) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.pipeline.stages.ExecutionFailureExceptionReportingStage.execute(ExecutionFailureExceptionReportingStage.java:26) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.http.AmazonSyncHttpClient$RequestExecutionBuilderImpl.execute(AmazonSyncHttpClient.java:193) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.invoke(BaseSyncClientHandler.java:128) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.doExecute(BaseSyncClientHandler.java:154) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.lambda$execute$1(BaseSyncClientHandler.java:107) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.measureApiCallSuccess(BaseSyncClientHandler.java:162) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.execute(BaseSyncClientHandler.java:91) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.core.client.handler.SdkSyncClientHandler.execute(SdkSyncClientHandler.java:45) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.awscore.client.handler.AwsSyncClientHandler.execute(AwsSyncClientHandler.java:55) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at shaded.software.amazon.awssdk.services.kinesis.DefaultKinesisClient.getRecords(DefaultKinesisClient.java:1050) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at org.apache.pinot.plugin.stream.kinesis.KinesisConsumer.getResult(KinesisConsumer.java:131) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at org.apache.pinot.plugin.stream.kinesis.KinesisConsumer.lambda$fetchMessages$0(KinesisConsumer.java:86) ~[pinot-kinesis-0.8.0-SNAPSHOT-shaded.jar:0.8.0-SNAPSHOT-78b08e4af41adc2e5e6d7387960df0524e2a7e90]
	at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:829) [?:?]

could some one please check. thanks

[mayankshriv]

Tagging @KKcorps @npawar.

[KKcorps]

Hi, The records from Kinesis are produced after every fetch timeout interval or when no new records are available in partition whichever comes first.
The above exception is just a warning when the thread is interrupted due to the fetch timeout and doesn't impact pinot kinesis processing. I will raise a PR to suppress this LOG.

[itsmesrds]

yea. other way is you can add little bigger timeout.

ClientOverrideConfiguration configuration = ClientOverrideConfiguration.builder()
.apiCallAttemptTimeout(Duration.ofMillis(1000))
.apiCallTimeout(Duration.ofMillis(1500))
.retryPolicy(RetryPolicy.defaultRetryPolicy())
.build();

according to the log, its creating new consumer after this warning. which may affect the performance.

[itsmesrds]

Hi @KKcorps .is this fix is also part of 0.8.0 release. its generating too much of log because of this.