abelyansky

public void processRecord(PutRecordsRequestEntry entry) {
   try {
       int recordSize = (int) this.kinesisClientManager.estimateSize(entry);

       // enforce size and rate limits by blocking until new tokens are generated
       entriesRatePerShard.acquire();
       bytesRatePerShard.acquire(recordSize);

       if (bufferQueue.size() + 1 >= MAX_ENTRIES_IN_REQUEST) {
           if (logger.isDebugEnabled()) logger.debug("buffer queue is at size limit. flushing");

abelyansky

import scala.concurrent.{Future,Await, blocking}
import scala.util.{Failure, Success}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

val WORKERS_WAIT_ON_THREADS = true

val (javaThreadCounter, scalaThreadCounter) = (sc.longAccumulator, sc.longAccumulator)

val df = sc.parallelize(1 to 10).toDF()

abelyansky

mime = require('mime')

wrk.method = "POST"
wrk.headers["Content-type"] = "application/x-protobuf"
requests = {}

init = function(args)
    io.input('/home/centos/cookies_small')
    for line in io.lines() do
      actor = mime.unb64(line)