Stream Transformer Failing to Fetch Records from Kinesis Stream

Daniel_Baron · August 10, 2022, 3:59pm

Hi

We have deployed the Snowplow Stream Transformer on an AWS ECS container and configured it to read from our Enriched Kinesis Data Stream.

We seem to get a lot of the following errors:

[prefetch-cache-shardId-000000000001-0000] [ERROR] software.amazon.kinesis.retrieval.polling.PrefetchRecordsPublisher - kinesis-production-service-spstreamenricher-enriched:shardId-000000000001 :  Exception thrown while fetching records from Kinesis

and…

[cats-effect-blocker-0] [ERROR] software.amazon.kinesis.lifecycle.ShardConsumerSubscriber - shardId-000000000001: Last request was dispatched at 2022-08-04T11:41:52.720721Z, but no response as of 2022-08-04T11:42:55.721381Z (PT1M3.00066S).  Cancelling subscription, and restarting. Last successful request details -- request id - NONE, timestamp - NONE

Our configuration looks like:

{
  "input": {
    "type": "kinesis",
    "appName": ${?DDB_NAME},
    "streamName": ${?INPUT_STREAM_NAME},
    "region": "eu-west-1",
    "position": "TRIM_HORIZON"
    "retrievalMode": {
      "type": "Polling"
      "maxRecords": 10000
    }
    "bufferSize": 3
  }
  "output": {
    "path": ${?TRANSFORMED_ARCHIVE_PATH},
    "compression": "GZIP",
    "region": "eu-west-1"
  }
  "windowing": "5 minutes"
  "queue": {
    "type": "sqs",
    "queueName": ${?SQS_QUEUE_NAME},
    "region": "${REGION}"
  }
  "formats": {
    "transformationType": "widerow",
    "default": "TSV",
  }
  "monitoring": {
    "metrics": {
      "stdout": {
        "period": "1 minute"
      }
      "cloudwatch": true
    }
  }
  "telemetry": {
    ...
  }
  "featureFlags": {
    "legacyMessageFormat": false
  }
}

When we push test events through nothing is written to S3 and the SQS queue remains empty. Anyone any ideas what is going on and what these error messages actually mean?

Thanks

BenB · August 16, 2022, 9:48am

Hi @Daniel_Baron ,

Could you share the lines right after these lines please, if any ?

This error comes from this line in KCL. It looks like the transformer sends a request to Kinesis but it takes too long to get an answer.

It looks correct.

The fact that not a single event gets written to S3 means that the transformer couldn’t consume any event from Kinesis. Either it couldn’t communicate with it at all, or it takes too long (but no response as of 2022-08-04T11:42:55.721381Z). It might have to do with the network configuration of your ECS container, could you share how it looks like please?

Daniel_Baron · August 16, 2022, 11:58am

Hi Ben,

Thanks for getting back to us on this!

Could you share the lines right after these lines please, if any ?

Please find the relevant log towards the top, following this are a number of stack traces (apologies for the volume!).

[cats-effect-blocker-0] [INFO] software.amazon.kinesis.coordinator.DiagnosticEventLogger - Current thread pool executor state: ExecutorStateEvent(executorName=SchedulerThreadPoolExecutor, currentQueueSize=0, activeThreads=0, coreThreads=0, leasesOwned=0, largestPoolSize=2, maximumPoolSize=2147483647)
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[prefetch-cache-shardId-000000000000-0000] [ERROR] software.amazon.kinesis.retrieval.polling.PrefetchRecordsPublisher - kinesis-production-service-spstreamenricher-enriched:shardId-000000000000 :  Exception thrown while fetching records from Kinesis
software.amazon.awssdk.core.exception.SdkClientException: Unable to execute HTTP request: Java heap space
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
\tat software.amazon.awssdk.core.exception.SdkClientException$BuilderImpl.build(SdkClientException.java:98)
\tat software.amazon.awssdk.core.exception.SdkClientException.create(SdkClientException.java:43)
\tat software.amazon.awssdk.core.internal.http.pipeline.stages.utils.RetryableStageHelper.setLastException(RetryableStageHelper.java:204)
\tat software.amazon.awssdk.core.internal.http.pipeline.stages.utils.RetryableStageHelper.setLastException(RetryableStageHelper.java:200)
\tat software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncRetryableStage$RetryingExecutor.maybeRetryExecute(AsyncRetryableStage.java:179)
\tat software.amazon.awssdk.core.internal.http.pipeline.stages.AsyncRetryableStage$RetryingExecutor.lambda$attemptExecute$1(AsyncRetryableStage.java:159)
\tat java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
\tat java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
\tat java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
\tat software.amazon.awssdk.utils.CompletableFutureUtils.lambda$forwardExceptionTo$0(CompletableFutureUtils.java:76)
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[pool-17-thread-1] [INFO] software.amazon.kinesis.leases.LeaseCleanupManager - Number of pending leases to clean before the scan : 0
\tat java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
\tat java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
\tat java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
\tat software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStage.lambda$null$0(MakeAsyncHttpRequestStage.java:104)
\tat java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
\tat java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
\tat java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
\tat software.amazon.awssdk.core.internal.http.pipeline.stages.MakeAsyncHttpRequestStage$WrappedErrorForwardingResponseHandler.onError(MakeAsyncHttpRequestStage.java:158)
\tat software.amazon.awssdk.http.nio.netty.internal.ResponseHandler$PublisherAdapter$1.lambda$notifyError$5(ResponseHandler.java:309)
\tat software.amazon.awssdk.http.nio.netty.internal.ResponseHandler.runAndLogError(ResponseHandler.java:181)
\tat software.amazon.awssdk.http.nio.netty.internal.ResponseHandler.access$500(ResponseHandler.java:71)
\tat software.amazon.awssdk.http.nio.netty.internal.ResponseHandler$PublisherAdapter$1.notifyError(ResponseHandler.java:307)
\tat software.amazon.awssdk.http.nio.netty.internal.utils.ExceptionHandlingUtils.tryCatch(ExceptionHandlingUtils.java:42)
\tat software.amazon.awssdk.http.nio.netty.internal.ResponseHandler$PublisherAdapter$1.onNext(ResponseHandler.java:270)
\tat software.amazon.awssdk.http.nio.netty.internal.ResponseHandler$PublisherAdapter$1.onNext(ResponseHandler.java:221)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.publishMessage(HandlerPublisher.java:407)
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.channelRead(HandlerPublisher.java:383)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler.handleReadHttpContent(HttpStreamsHandler.java:228)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler.channelRead(HttpStreamsHandler.java:199)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsClientHandler.channelRead(HttpStreamsClientHandler.java:173)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
\tat software.amazon.awssdk.http.nio.netty.internal.LastHttpContentHandler.channelRead(LastHttpContentHandler.java:43)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
\tat software.amazon.awssdk.http.nio.netty.internal.http2.Http2ToHttpInboundAdapter.onDataRead(Http2ToHttpInboundAdapter.java:84)
\tat software.amazon.awssdk.http.nio.netty.internal.http2.Http2ToHttpInboundAdapter.channelRead0(Http2ToHttpInboundAdapter.java:49)
\tat software.amazon.awssdk.http.nio.netty.internal.http2.Http2ToHttpInboundAdapter.channelRead0(Http2ToHttpInboundAdapter.java:42)
\tat io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
\tat io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
\tat io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
\tat io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.doRead0(AbstractHttp2StreamChannel.java:901)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.doBeginRead(AbstractHttp2StreamChannel.java:816)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.beginRead(AbstractHttp2StreamChannel.java:785)
\tat io.netty.channel.DefaultChannelPipeline$HeadContext.read(DefaultChannelPipeline.java:1362)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
\tat io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
\tat software.amazon.awssdk.http.nio.netty.internal.http2.FlushOnReadHandler.read(FlushOnReadHandler.java:40)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
\tat io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.requestDemand(HandlerPublisher.java:130)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler$1.requestDemand(HttpStreamsHandler.java:191)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.channelReadComplete(HandlerPublisher.java:424)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler.channelReadComplete(HttpStreamsHandler.java:248)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsClientHandler.channelReadComplete(HttpStreamsClientHandler.java:59)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat io.netty.handler.timeout.IdleStateHandler.channelReadComplete(IdleStateHandler.java:295)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat io.netty.channel.DefaultChannelPipeline$HeadContext.channelReadComplete(DefaultChannelPipeline.java:1415)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.DefaultChannelPipeline.fireChannelReadComplete(DefaultChannelPipeline.java:925)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.notifyReadComplete(AbstractHttp2StreamChannel.java:871)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.doBeginRead(AbstractHttp2StreamChannel.java:827)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.beginRead(AbstractHttp2StreamChannel.java:785)
\tat io.netty.channel.DefaultChannelPipeline$HeadContext.read(DefaultChannelPipeline.java:1362)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
\tat io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
\tat software.amazon.awssdk.http.nio.netty.internal.http2.FlushOnReadHandler.read(FlushOnReadHandler.java:40)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
\tat io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.requestDemand(HandlerPublisher.java:130)
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler$1.requestDemand(HttpStreamsHandler.java:191)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.channelReadComplete(HandlerPublisher.java:424)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler.channelReadComplete(HttpStreamsHandler.java:248)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsClientHandler.channelReadComplete(HttpStreamsClientHandler.java:59)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat io.netty.handler.timeout.IdleStateHandler.channelReadComplete(IdleStateHandler.java:295)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat io.netty.channel.DefaultChannelPipeline$HeadContext.channelReadComplete(DefaultChannelPipeline.java:1415)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.DefaultChannelPipeline.fireChannelReadComplete(DefaultChannelPipeline.java:925)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.notifyReadComplete(AbstractHttp2StreamChannel.java:871)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.doBeginRead(AbstractHttp2StreamChannel.java:827)
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.beginRead(AbstractHttp2StreamChannel.java:785)
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
\tat io.netty.channel.DefaultChannelPipeline$HeadContext.read(DefaultChannelPipeline.java:1362)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
\tat io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
\tat software.amazon.awssdk.http.nio.netty.internal.http2.FlushOnReadHandler.read(FlushOnReadHandler.java:40)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
\tat io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.requestDemand(HandlerPublisher.java:130)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler$1.requestDemand(HttpStreamsHandler.java:191)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.channelReadComplete(HandlerPublisher.java:424)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler.channelReadComplete(HttpStreamsHandler.java:248)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsClientHandler.channelReadComplete(HttpStreamsClientHandler.java:59)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat io.netty.handler.timeout.IdleStateHandler.channelReadComplete(IdleStateHandler.java:295)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.AbstractChannelHandlerContext.fireChannelReadComplete(AbstractChannelHandlerContext.java:390)
\tat io.netty.channel.DefaultChannelPipeline$HeadContext.channelReadComplete(DefaultChannelPipeline.java:1415)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:410)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeChannelReadComplete(AbstractChannelHandlerContext.java:397)
\tat io.netty.channel.DefaultChannelPipeline.fireChannelReadComplete(DefaultChannelPipeline.java:925)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.notifyReadComplete(AbstractHttp2StreamChannel.java:871)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.doBeginRead(AbstractHttp2StreamChannel.java:827)
\tat io.netty.handler.codec.http2.AbstractHttp2StreamChannel$Http2ChannelUnsafe.beginRead(AbstractHttp2StreamChannel.java:785)
\tat io.netty.channel.DefaultChannelPipeline$HeadContext.read(DefaultChannelPipeline.java:1362)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
\tat io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
\tat software.amazon.awssdk.http.nio.netty.internal.http2.FlushOnReadHandler.read(FlushOnReadHandler.java:40)
\tat io.netty.channel.AbstractChannelHandlerContext.invokeRead(AbstractChannelHandlerContext.java:686)
\tat io.netty.channel.AbstractChannelHandlerContext.read(AbstractChannelHandlerContext.java:671)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.requestDemand(HandlerPublisher.java:130)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsHandler$1.requestDemand(HttpStreamsHandler.java:191)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.flushBuffer(HandlerPublisher.java:345)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.receivedDemand(HandlerPublisher.java:291)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher.access$200(HandlerPublisher.java:61)
\tat software.amazon.awssdk.http.nio.netty.internal.nrs.HandlerPublisher$ChannelSubscription$1.run(HandlerPublisher.java:495)
\tat io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
\tat io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
[cats-effect-blocker-0] [INFO] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - getCurrentlyHeldLease not returning lease with key shardId-000000000001 because it is expired
\tat io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:497)
\tat io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
\tat io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
\tat java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.OutOfMemoryError: Java heap space

It’s worth noting that we’ve made 2 changes that seem to have resolved this:

Set the retrievalMode to FanOut
6x the available memory for the task

The fact that not a single event gets written to S3 means that the transformer couldn’t consume any event from Kinesis. Either it couldn’t communicate with it at all, or it takes too long (but no response as of 2022-08-04T11:42:55.721381Z ). It might have to do with the network configuration of your ECS container, could you share how it looks like please?

Given the above (we are now seeing messages in the SQS queue and data in S3) I think the networking must be okay? But this seems consistent with the lack of memory, I’ve also noticed some java heap space exceptions (example in the logs above ^).

Would be good to fully understand what’s these errors mean for future reference (even if it’s just a memory issue) before we close this.

Thanks again for getting back to us!
Dan

Daniel_Baron · August 17, 2022, 9:31am

Worth also noting we have since seen the following errors too:

17/08/2022 08:28:49.0048+0000 [LeaseCoordinator-0001] [ERROR] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseRenewer - Encountered an exception while renewing a lease.
17/08/2022 08:28:49.0050+0000 [LeaseCoordinator-0001] [ERROR] software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseCoordinator - LeasingException encountered in lease renewing thread

These look to be related to the Kinesis Client and the suggested fix is to upgrade. Any way we can solve this. We are already running the 4.2.1 transformer?

Thanks again!

BenB · August 19, 2022, 2:35pm

Hi @Daniel_Baron,

Glad to hear that you could make some progress on this.

Just to get more context, what is the throughput of your pipeline? I’d like to see if we are talking about a low-volume pipeline or not.

I wonder if you could be reaching DymanoDB quotas, thus explaining the leases that fail to renew, and also the error from your other message, could you check in Cloudwatch metrics if that could be the case please ? Although it might be related to the OOM issue.

You might be holding too many events in-memory at once.

You could try to decrease maxRecords to something smaller (e.g. 1000).

I doubt that switching to FanOut has an impact, so it must be the memory. How much was allocated before?

Indeed networking must be fine then!

Please try reducing maxRecords and let us know what you observe.

Topic		Replies	Views
Unable to receive Snowplow data into Elasticsearch Data store sources	14	3416	January 17, 2018
Scala Stream Collector and Kinesis Shards AWS real-time pipeline	1	1790	September 13, 2017
Kinesis Stream Transformer not writing to AWS SQS For engineers	2	831	August 3, 2022
Kinesis stream enrich failing - TimeoutException AWS real-time pipeline	8	1349	February 19, 2021
Elasticsearch sink - Application processRecords() threw an exception when processing shard Storage targets	10	2752	July 18, 2017

Stream Transformer Failing to Fetch Records from Kinesis Stream

Related topics