[
https://issues.apache.org/jira/browse/NIFI-15319?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Alaksiej Ščarbaty updated NIFI-15319:
-------------------------------------
Description:
`ConsumeKinesis` processor doesn't accept any new records due to the buffer
being full.
However, when reading records from the buffer no records are received.
The reason:
This is caused by not decreasing the memory counter when a [consumer is
shutdown|https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/ShardRecordProcessor.java#L78].
Eventually causing the memory counter to reach the max allowed user memory,
while in reality no data is buffered.In the [shard processor
docs|https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/ShardRecordProcessor.java#L70]
it's claimed that consumers are shut down when Scheduler is shut down. The
same is true for [the shutdown
reason|https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownReason.java#L49].
That's why the memory tracker was not adjusted, since the whole processor is
stopping anyway.But this wasn't true. In the logs (and then [in the
code|https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/LeaseGracefulShutdownHandler.java#L135])
I noticed that a single consumer can be shutdown when during a graceful lease
handoff.
The proposed fix:
When a buffer is shut down doi the invalidation, and decrease memory counter.
[Code
ref|https://github.com/apache/nifi/blob/main/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/MemoryBoundRecordBuffer.java#L588].
was:
`ConsumeKinesis` processor doesn't accept any new records due to the buffer
being full.
However, when reading records from the buffer no records are received.
> ConsumeKinesis throttles new records, but buffers are empty
> -----------------------------------------------------------
>
> Key: NIFI-15319
> URL: https://issues.apache.org/jira/browse/NIFI-15319
> Project: Apache NiFi
> Issue Type: Bug
> Reporter: Alaksiej Ščarbaty
> Assignee: Alaksiej Ščarbaty
> Priority: Major
>
> `ConsumeKinesis` processor doesn't accept any new records due to the buffer
> being full.
> However, when reading records from the buffer no records are received.
>
> The reason:
> This is caused by not decreasing the memory counter when a [consumer is
> shutdown|https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/ShardRecordProcessor.java#L78].
> Eventually causing the memory counter to reach the max allowed user memory,
> while in reality no data is buffered.In the [shard processor
> docs|https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/ShardRecordProcessor.java#L70]
> it's claimed that consumers are shut down when Scheduler is shut down. The
> same is true for [the shutdown
> reason|https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownReason.java#L49].
> That's why the memory tracker was not adjusted, since the whole processor is
> stopping anyway.But this wasn't true. In the logs (and then [in the
> code|https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/LeaseGracefulShutdownHandler.java#L135])
> I noticed that a single consumer can be shutdown when during a graceful
> lease handoff.
>
> The proposed fix:
> When a buffer is shut down doi the invalidation, and decrease memory counter.
> [Code
> ref|https://github.com/apache/nifi/blob/main/nifi-extension-bundles/nifi-aws-bundle/nifi-aws-kinesis/src/main/java/org/apache/nifi/processors/aws/kinesis/MemoryBoundRecordBuffer.java#L588].
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
