Styjun

Problem: initialize(...) method is called multiple times on a recordProcessor when that recordProcessor has high count of children threads.

Environment:

AWS SDK: 1.11.498

KCL: 1.9.3

Scala: 2.12.8

JDK: OpenJDK 1.8

Host: Amazon Linux 2 on ECS [ami-007571470797b8ffa]

Number of shards: 8

Implementation:

I have implemented a RecordProcessor (extending IRecordProcessor) in Scala.
- This record processor starts a stats reporting thread on initialize(...) to report stats to our collectors.
- Also, this record processor internally distributes records to be processed among multiple worker threads via a work queue when processRecords(...) is called. These worker threads are also started on initialize(...).

Problem:
When shard count is 8 and worker threads count is 16, KCL calls initialize(...) method multiple times on same recordProcessor. This throws IllegalThreadStateException as statsReporterThread is tried to be started when it's already started by previous initialize(...) call.

Catch:
When shard count is still 8, but worker thread count is 1, multiple initialize(...) calls are NOT made and everything works perfect.
This is puzzling as worker threads are not exposed to KCL, they are internal implementation of the record processor.
I suspected lower ulimits, so I increased them, it didn't help.

Also, when this same application is run on my laptop, it works! But fails on AWS ECS.

Code:

class RecordProcessor() extends IRecordProcessor 

 val statsReporter = new StatsReporter()
 val statsReporterThread = new Thread(statsReporter)

 val workQueue: LinkedBlockingDeque[Record] = ...
 val workerThreads: ListBuffer[Thread] = ...

 def initialize(shardId) = 
 statsReporterThread.start()

 (0 until 16).foreach(_ => 
 val wThread = new Thread(new Worker(workQueue))
 workerThreads += wThread
 wThread.start()
 )
 

 def processRecords(records, checkpointer) = 
 records.foreach(record => 
 wq.put(record)
 )

 if (currentTimeMs > nextTimeInMs) 
 checkpoint(checkpointer)
 nextTimeInMs = currentTimeMs + 15000
 
 

 def shutdown(checkpointer, reason) = 
 workerThreads.foreach(w => 
 w.interrupt()
 w.join()
 )

 statsReporterThread.interrupt()
 statsReporterThread.join()
 


class Worker(workQueue) extends Runnable 
 override def run(): Unit = 
 while (!Thread.currentThread().isInterrupted) 
 val record = q.take()
 process(record)
 
 

Any help / pointers would be highly appreciated!

Thanks!