[
https://issues.apache.org/jira/browse/FLINK-4536?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15457533#comment-15457533
]
Tzu-Li (Gordon) Tai edited comment on FLINK-4536 at 9/2/16 4:58 AM:
--------------------------------------------------------------------
Hi [~skidder], just to clarify, the thread leak issue should be resolved after
the fix on your InfluxDB Sink, and like the discussion above, should be
independent of the Kinesis connector, correct?
The cached thread pool created by the Kinesis connector is also irrelevant to
the AWS SDK and KCL version used, and didn't seem to be causing any thread
leaks. So, I would suggest that this particular issue is resolved ;)
As for the ProvisionedThroughputExceeded failures:
The connector fails after hitting 3 consecutive ProvisionedThroughputExceeded
exceptions. This is an AWS service usage limitation
(http://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html).
The normal way to tune this is by setting
{{ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS}} to a higher value
(default is 0 interval, at most 1000ms should be sufficient for most cases) in
the consumer's configuration properties, so that the consumer subtasks don't
fetch records that frequently, and accordingly also increase
{{ConsumerConfigConstants.SHARD_GETRECORDS_MAX}} to let Kinesis return more
records on each fetch.
If you are suggesting that the AWS SDK version bump to 1.11.30 somehow lessens
encountering the ProvisionedThroughputExceeded, could you open a separate JIRA
for this and we continue any discussion on this over there? As for the KCL, it
should be irrelevant to the issues you were facing; we're only using a single
utillity class from it to deaggregate Kinesis records.
was (Author: tzulitai):
Hi [~skidder], just to clarify, the thread leak issue should be resolved after
the fix on your InfluxDB Sink, and like the discussion above, should be
independent of the Kinesis connector, correct?
The cached thread pool created by the Kinesis connector is also irrelevant to
the AWS SDK and KCL version used. So, I would suggest that this particular
issue is resolved ;)
As for the ProvisionedThroughputExceeded failures:
The connector fails after hitting 3 consecutive ProvisionedThroughputExceeded
exceptions. This is an AWS service usage limitation
(http://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html).
The normal way to tune this is by setting
{{ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS}} to a higher value
(default is 0 interval, at most 1000ms should be sufficient for most cases) in
the consumer's configuration properties, so that the consumer subtasks don't
fetch records that frequently, and accordingly also increase
{{ConsumerConfigConstants.SHARD_GETRECORDS_MAX}} to let Kinesis return more
records on each fetch.
If you are suggesting that the AWS SDK version bump to 1.11.30 somehow lessens
encountering the ProvisionedThroughputExceeded, could you open a separate JIRA
for this and we continue any discussion on this over there? As for the KCL, it
should be irrelevant to the issues you were facing; we're only using a single
utillity class from it to deaggregate Kinesis records.
> Possible thread leak in Task Manager
> ------------------------------------
>
> Key: FLINK-4536
> URL: https://issues.apache.org/jira/browse/FLINK-4536
> Project: Flink
> Issue Type: Bug
> Components: TaskManager
> Affects Versions: 1.1.0
> Reporter: Scott Kidder
>
> Running Flink release 1.1.1 commit 61bfb36 in the following configuration:
> Job Manager
> 2 x Task Manager (2 CPU cores on each Task Manager)
> I've also updated the Kinesis source to use the latest AWS Java SDK, release
> 1.11.29.
> I've got a single Flink application using all 4 slots. It consumes from a
> Kinesis stream configured with 2 shards. I've limited the Kinesis source to a
> parallelism of 2 as a workaround for FLINK-4341.
> Occasionally the Kinesis consumer fails because of provisioned-throughput
> limits being hit. The application automatically restarts, and resumes
> processing with the checkpoint stored on the Job Manager with no outward
> indication of problems.
> I recently enabled the StatsD metrics reporter in Flink and noticed that the
> number of threads running on each Task Manager increases by about 20 threads
> each time the application restarts. Over the course of a day the application
> might hit provisioned-throughput limits 20 times or so (this is not fully
> production yet, so hitting these limits is acceptable for now). But the
> number of threads continues to grow unbounded with no increase in workload on
> the Task Managers.
> The following link includes charts for the overall Flink cluster performance
> & Task Manager JVM threads over the course of 12 hours:
> http://imgur.com/a/K59hz
> Each decrease and subsequent spike in threads corresponds to the job being
> restarted due to an AWS Kinesis source error.
> Here are the logs from one of the Task Manager instances on startup:
> {code}
> 2016-08-30 14:52:50,438 WARN org.apache.hadoop.util.NativeCodeLoader
> - Unable to load native-hadoop library for your platform...
> using builtin-java classes where applicable
> 2016-08-30 14:52:50,540 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> --------------------------------------------------------------------------------
> 2016-08-30 14:52:50,540 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Starting
> TaskManager (Version: 1.1.1, Rev:61bfb36, Date:09.08.2016 @ 12:09:08 UTC)
> 2016-08-30 14:52:50,540 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Current
> user: root
> 2016-08-30 14:52:50,541 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - JVM: OpenJDK
> 64-Bit Server VM - Oracle Corporation - 1.8/25.92-b14
> 2016-08-30 14:52:50,541 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Maximum heap
> size: 2048 MiBytes
> 2016-08-30 14:52:50,541 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - JAVA_HOME:
> /usr/lib/jvm/java-1.8-openjdk/jre
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Hadoop
> version: 2.7.2
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - JVM Options:
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> -XX:+UseG1GC
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - -Xms2048M
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - -Xmx2048M
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> -XX:MaxDirectMemorySize=8388607T
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> -Dlog.file=/usr/local/flink-1.1.1/log/flink--taskmanager-1-ip-10-55-2-218.log
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> -Dlog4j.configuration=file:/usr/local/flink-1.1.1/conf/log4j.properties
> 2016-08-30 14:52:50,543 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> -Dlogback.configurationFile=file:/usr/local/flink-1.1.1/conf/logback.xml
> 2016-08-30 14:52:50,544 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Program
> Arguments:
> 2016-08-30 14:52:50,544 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> --configDir
> 2016-08-30 14:52:50,544 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> /usr/local/flink-1.1.1/conf
> 2016-08-30 14:52:50,544 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Classpath:
> /usr/local/flink-1.1.1/lib/flink-dist_2.11-1.1.1.jar:/usr/local/flink-1.1.1/lib/flink-metrics-statsd-1.1.1.jar:/usr/local/flink-1.1.1/lib/flink-python_2.11-1.1.1.jar:/usr/local/flink-1.1.1/lib/log4j-1.2.17.jar:/usr/local/flink-1.1.1/lib/slf4j-log4j12-1.7.7.jar:::
> 2016-08-30 14:52:50,544 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> --------------------------------------------------------------------------------
> 2016-08-30 14:52:50,544 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Registered
> UNIX signal handlers for [TERM, HUP, INT]
> 2016-08-30 14:52:50,547 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Maximum
> number of open file descriptors is 1048576
> 2016-08-30 14:52:50,565 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Loading
> configuration from /usr/local/flink-1.1.1/conf
> 2016-08-30 14:52:50,610 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Security is
> not enabled. Starting non-authenticated TaskManager.
> 2016-08-30 14:52:50,610 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Using
> configured hostname/address for TaskManager: 10.55.2.218
> 2016-08-30 14:52:50,611 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Starting
> TaskManager
> 2016-08-30 14:52:50,615 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Starting
> TaskManager actor system at 10.55.2.218:0
> 2016-08-30 14:52:50,956 INFO akka.event.slf4j.Slf4jLogger
> - Slf4jLogger started
> 2016-08-30 14:52:51,005 INFO Remoting
> - Starting remoting
> 2016-08-30 14:52:51,159 INFO Remoting
> - Remoting started; listening on addresses
> :[akka.tcp://flink@10.55.2.218:44007]
> 2016-08-30 14:52:51,163 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Starting
> TaskManager actor
> 2016-08-30 14:52:51,177 INFO
> org.apache.flink.runtime.io.network.netty.NettyConfig - NettyConfig
> [server address: /10.55.2.218, server port: 36007, memory segment size
> (bytes): 32768, transport type: NIO, number of server threads: 2 (manual),
> number of client threads: 2 (manual), server connect backlog: 0 (use Netty's
> default), client connect timeout (sec): 120, send/receive buffer size
> (bytes): 0 (use Netty's default)]
> 2016-08-30 14:52:51,179 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Messages
> between TaskManager and JobManager have a max timeout of 10000 milliseconds
> 2016-08-30 14:52:51,183 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Temporary
> file directory '/tmp': total 9 GB, usable 8 GB (88.89% usable)
> 2016-08-30 14:52:51,227 INFO
> org.apache.flink.runtime.io.network.buffer.NetworkBufferPool - Allocated 64
> MB for network buffer pool (number of memory segments: 2048, bytes per
> segment: 32768).
> 2016-08-30 14:52:51,287 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Limiting
> managed memory to 0.7 of the currently free heap space (1377 MB), memory will
> be allocated lazily.
> 2016-08-30 14:52:51,309 INFO
> org.apache.flink.runtime.io.disk.iomanager.IOManager - I/O manager
> uses directory /tmp/flink-io-c000b71a-d793-4f4c-90c4-0c0808154219 for spill
> files.
> 2016-08-30 14:52:51,325 INFO org.apache.flink.runtime.filecache.FileCache
> - User file cache uses directory
> /tmp/flink-dist-cache-95fd32d3-a629-42ce-816e-ca8a63e13c7d
> 2016-08-30 14:52:51,529 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Starting
> TaskManager actor at akka://flink/user/taskmanager#496823118.
> 2016-08-30 14:52:51,529 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - TaskManager
> data connection information: ip-10-55-2-218.ec2.internal (dataPort=36007)
> 2016-08-30 14:52:51,530 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - TaskManager
> has 2 task slot(s).
> 2016-08-30 14:52:51,531 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Memory usage
> stats: [HEAP: 93/2048/2048 MB, NON HEAP: 31/32/-1 MB (used/committed/max)]
> 2016-08-30 14:52:51,534 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Trying to
> register at JobManager akka.tcp://flink@10.55.2.212:6123/user/jobmanager
> (attempt 1, timeout: 500 milliseconds)
> 2016-08-30 14:52:51,690 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Successful
> registration at JobManager
> (akka.tcp://flink@10.55.2.212:6123/user/jobmanager), starting network stack
> and library cache.
> 2016-08-30 14:52:51,841 INFO
> org.apache.flink.runtime.io.network.netty.NettyClient - Successful
> initialization (took 28 ms).
> 2016-08-30 14:52:51,871 INFO
> org.apache.flink.runtime.io.network.netty.NettyServer - Successful
> initialization (took 30 ms). Listening on SocketAddress /10.55.2.218:36007.
> 2016-08-30 14:52:51,872 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Determined
> BLOB server address to be /10.55.2.212:55892. Starting BLOB cache.
> 2016-08-30 14:52:51,874 INFO org.apache.flink.runtime.blob.BlobCache
> - Created BLOB cache storage directory
> /tmp/blobStore-81d7d33f-a815-4f0b-9131-d3ba5a256d1b
> 2016-08-30 14:52:51,889 INFO org.apache.flink.metrics.statsd.StatsDReporter
> - Starting StatsDReporter to send metric reports to
> localhost/127.0.0.1:8125
> 2016-08-30 14:52:51,891 INFO org.apache.flink.runtime.metrics.MetricRegistry
> - Periodically reporting metrics in intervals of 10 SECONDS for
> reporter statsd of type org.apache.flink.metrics.statsd.StatsDReporter.
> {code}
> The Kinesis reader errors look like:
> {code}
> 2016-08-30 17:23:43,353 WARN
> org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy - Got
> ProvisionedThroughputExceededException. Backing off for 53 millis.
> 2016-08-30 17:23:43,725 WARN
> org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy - Got
> ProvisionedThroughputExceededException. Backing off for 597 millis.
> 2016-08-30 17:23:44,805 WARN
> org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy - Got
> ProvisionedThroughputExceededException. Backing off for 538 millis.
> 2016-08-30 17:23:45,344 INFO
> org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
> Shutting down the shard consumer threads of subtask 0 ...
> 2016-08-30 17:23:45,394 INFO
> org.apache.flink.streaming.runtime.tasks.StreamTask - Timer service
> is shutting down.
> 2016-08-30 17:23:45,395 INFO
> org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher -
> Shutting down the shard consumer threads of subtask 0 ...
> 2016-08-30 17:23:45,395 ERROR org.apache.flink.runtime.taskmanager.Task
> - Task execution failed.
> java.lang.RuntimeException: Rate Exceeded for getRecords operation - all
> 3retryattempts returned ProvisionedThroughputExceededException.
> at
> org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy.getRecords(KinesisProxy.java:204)
> at
> org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.run(ShardConsumer.java:167)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> 2016-08-30 17:23:45,396 INFO org.apache.flink.runtime.taskmanager.Task
> - Source: Kinesis (1/2) switched to FAILED with exception.
> java.lang.RuntimeException: Rate Exceeded for getRecords operation - all
> 3retryattempts returned ProvisionedThroughputExceededException.
> at
> org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy.getRecords(KinesisProxy.java:204)
> at
> org.apache.flink.streaming.connectors.kinesis.internals.ShardConsumer.run(ShardConsumer.java:167)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> 2016-08-30 17:23:45,396 INFO org.apache.flink.runtime.taskmanager.Task
> - Freeing task resources for Source: Kinesis (1/2)
> 2016-08-30 17:23:45,396 INFO
> org.apache.flink.streaming.runtime.tasks.StreamTask - Timer service
> is shutting down.
> 2016-08-30 17:23:45,396 INFO
> org.apache.flink.runtime.taskmanager.TaskManager -
> Un-registering task and sending final execution state FAILED to JobManager
> for task Source: Kinesis (c1bf869b9d8506cea67b1317b21c014e)
> 2016-08-30 17:23:45,396 INFO
> org.apache.flink.streaming.runtime.tasks.StreamTask - Timer service
> is shutting down.
> 2016-08-30 17:23:45,402 INFO
> org.apache.flink.runtime.taskmanager.TaskManager - Discarding
> the results produced by task execution c1bf869b9d8506cea67b1317b21c014e
> 2016-08-30 17:23:45,403 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Parse -> Timestamp and Watermark ->
> (Map, Map, Map, Map) (1/4)
> 2016-08-30 17:23:45,403 INFO org.apache.flink.runtime.taskmanager.Task
> - Parse -> Timestamp and Watermark -> (Map, Map, Map, Map)
> (1/4) switched to CANCELING
> 2016-08-30 17:23:45,403 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Parse -> Timestamp and
> Watermark -> (Map, Map, Map, Map) (1/4) (70809fea05a6f67ffd7672bbe5b9643a).
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Parse -> Timestamp and Watermark ->
> (Map, Map, Map, Map) (3/4)
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Parse -> Timestamp and Watermark -> (Map, Map, Map, Map)
> (3/4) switched to CANCELING
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Parse -> Timestamp and
> Watermark -> (Map, Map, Map, Map) (3/4) (a70eafedd01af499a4ee066108e92ae8).
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Fold: property_id, video_id ->
> 1-minute Fixed-Window Percentile Aggregation -> Sink: InfluxDB (1/4)
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Fold: property_id, video_id -> 1-minute Fixed-Window
> Percentile Aggregation -> Sink: InfluxDB (1/4) switched to CANCELING
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Fold: property_id,
> video_id -> 1-minute Fixed-Window Percentile Aggregation -> Sink: InfluxDB
> (1/4) (b71730eb5cd484cde0eb8332e69d443e).
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Fold: property_id, video_id ->
> 1-minute Fixed-Window Percentile Aggregation -> Sink: InfluxDB (2/4)
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Fold: property_id, video_id -> 1-minute Fixed-Window
> Percentile Aggregation -> Sink: InfluxDB (2/4) switched to CANCELING
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Fold: property_id,
> video_id -> 1-minute Fixed-Window Percentile Aggregation -> Sink: InfluxDB
> (2/4) (6115f675c4d36004d4c885c9868d6b61).
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Fold: property_id, video_id ->
> 5-minute Sliding-Window Percentile Aggregation -> Sink: InfluxDB (1/4)
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Fold: property_id, video_id -> 5-minute Sliding-Window
> Percentile Aggregation -> Sink: InfluxDB (1/4) switched to CANCELING
> 2016-08-30 17:23:45,405 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Fold: property_id,
> video_id -> 5-minute Sliding-Window Percentile Aggregation -> Sink: InfluxDB
> (1/4) (d9cb8238533083dc2396e90cae3e702e).
> 2016-08-30 17:23:45,406 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Fold: property_id, video_id ->
> 5-minute Sliding-Window Percentile Aggregation -> Sink: InfluxDB (2/4)
> 2016-08-30 17:23:45,406 INFO org.apache.flink.runtime.taskmanager.Task
> - Fold: property_id, video_id -> 5-minute Sliding-Window
> Percentile Aggregation -> Sink: InfluxDB (2/4) switched to CANCELING
> 2016-08-30 17:23:45,406 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Fold: property_id,
> video_id -> 5-minute Sliding-Window Percentile Aggregation -> Sink: InfluxDB
> (2/4) (311ffa604444b33985572fe99170f405).
> 2016-08-30 17:23:45,406 INFO
> org.apache.flink.streaming.runtime.tasks.StreamTask - Timer service
> is shutting down.
> 2016-08-30 17:23:45,406 INFO org.apache.flink.runtime.taskmanager.Task
> - Attempting to cancel task Fold: property_id, video_id ->
> 10-minute Sliding-Window Percentile Aggregation -> Sink: InfluxDB (1/4)
> 2016-08-30 17:23:45,420 INFO org.apache.flink.runtime.taskmanager.Task
> - Fold: property_id, video_id -> 10-minute Sliding-Window
> Percentile Aggregation -> Sink: InfluxDB (1/4) switched to CANCELING
> 2016-08-30 17:23:45,420 INFO org.apache.flink.runtime.taskmanager.Task
> - Triggering cancellation of task code Fold: property_id,
> video_id -> 10-minute Sliding-Window Percentile Aggregation -> Sink: InfluxDB
> (1/4) (556a4e6d2a4b7686539b50ad6b5f0d0d).
> 2016-08-30 17:23:45,420 INFO
> org.apache.flink.streaming.runtime.tasks.StreamTask - Timer service
> is shutting down.
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
