Adding BatchSizeManager to Castle #3

shubhrojroy · 2017-05-03T01:26:33Z

Adding BatchSizeManager to Castle to control bytes read during fetch.

denisgrenader · 2017-05-03T02:27:33Z

src/main/scala/com/box/castle/core/config/CommitterConfig.scala

@@ -75,4 +80,7 @@ object CommitterConfig {
  val DefaultParallelismFactorByTopic = Map.empty[String, Int]
  val DefaultCorruptMessagePolicy = CorruptMessagePolicy.skip
  val DefaultUseKafkaMetadataManager = true
+  val DefaultUseBatchSizeManager = false
+  val DefaultSamplingSlots = 20
+  val DefaultSamplingInterval = 60000 // milliseconds


this should be of type Duration

denisgrenader · 2017-05-03T02:29:57Z

src/main/scala/com/box/castle/core/common/BoundedQueue.scala

+/**
+  * Queue with a fixed upper bound on capacity
+  */
+class BoundedQueue[T](capacity: Int) extends mutable.Queue[T] {


this class needs its own unit tests

denisgrenader · 2017-05-03T02:34:20Z

src/main/scala/com/box/castle/core/committer/manager/BatchSizeManager.scala

+
+object BatchSizeManager {
+  val DiscountFactor = 0.9
+  val FullBufferMaxCount = 3


DiscountFactor & FullBufferMaxCount should be passed in via config, so you can remove this companion object

denisgrenader · 2017-05-03T02:36:55Z

src/main/scala/com/box/castle/core/committer/manager/BatchSizeManager.scala

+      // Update number of times we have seen consecutive full buffer reads
+      consecutiveFullBuffers += 1
+    }
+    else if (consecutiveFullBuffers > 0) {


no need for else if check here, just else { consecutiveFullBuffers = 0 }

denisgrenader · 2017-05-03T02:37:59Z

src/main/scala/com/box/castle/core/committer/manager/BatchSizeManager.scala

+    }
+
+    if (consecutiveFullBuffers > FullBufferMaxCount) {
+      // We are mostly in catchup mode


// We are most likely in catchup mode

denisgrenader · 2017-05-03T02:58:52Z

src/main/scala/com/box/castle/core/committer/states/CommittingBatch.scala

@@ -118,7 +123,11 @@ trait CommittingBatch extends CommitterActorBase
    // We commit the consumer offset here without waiting on the result, we are assuming it will succeed most of the time
    commitConsumerOffset(batch.nextOffset, metadata)

-    becomeFetchingData(OffsetAndMetadata(batch.nextOffset, metadata))
+    // If BatchSizeManager is enabled then goto Idling state with a specific delay
+    if(committerConfig.useBatchSizeManager)


due to Option[BatchSizeManager] this becomes a match case statement

denisgrenader · 2017-05-03T02:59:38Z

src/main/scala/com/box/castle/core/committer/states/CommittingBatch.scala

@@ -153,6 +162,10 @@ trait CommittingBatch extends CommitterActorBase
    // We fetch the latest offset in the topic here so we can compute the offset lag
    sendRequestToRouter(FetchOffset(LatestOffset, topicAndPartition))

+    // Track bytes read from Kafka if BatchSizeManager is enabled
+    if(committerConfig.useBatchSizeManager)


due to Option[BatchSizeManager] you can make this into a batchSizeManager.foreach(_.track(batch.sizeInBytes, System.currentTimeMillis())) call instead of an if statement

denisgrenader · 2017-05-03T03:03:32Z

src/test/scala/com/box/castle/core/committer/manager/BatchSizeManagerTest.scala

+      batchSizeManager.track(bytesRead, ts2)
+
+      // Validate
+      val expectedResult = 100 * 0.9 / (20 / (ts2 - ts1))


with 3 samples of 10 bytes read, means you have read 300 bytes in 300ms, your rate here is 1 byte / ms, if your buffer size is 1200, then expected should be (1200 / 1) * 0.9

Not sure I get what you mean here.

denisgrenader · 2017-05-03T03:04:03Z

src/test/scala/com/box/castle/core/committer/manager/BatchSizeManagerTest.scala

+      // Setup
+      val samplingSlots = 3
+      val samplingInterval = 100 // milliseconds
+      val bufferSize = 100 //bytes


it's very confusing in a test to make buffer size equal to sampling interval, can you change it to any other number that looks different like 1200

denisgrenader · 2017-05-03T03:06:59Z

src/test/scala/com/box/castle/core/committer/manager/BatchSizeManagerTest.scala

+      // Execute
+      val ts1 = System.currentTimeMillis()
+      batchSizeManager.track(bytesRead, ts1)
+      Thread.sleep(samplingInterval)


these tests will be super flaky, you cannot Thread.sleep() here. mock the call to System.currentTimeMillis(), or... my personal favorite is just create your own method for getting time in your BatchSizeManager and just mock that method with the value you want it to return

denisgrenader · 2017-05-03T03:18:00Z

src/main/scala/com/box/castle/core/committer/manager/BatchSizeManager.scala

+      consecutiveFullBuffers = 0
+    }
+
+    if (samples.isEmpty || timestamp - samples.last.timestamp > samplingInterval) {


we need to fill the samples as quickly as possible before we start dropping them due to the sampling interval that is desired, so this should be: if (!(samples.isFull) || (timestamp - samples.last.timestamp) >= samplingInterval)

shubhrojroy · 2017-05-03T22:31:14Z

Made changes to address comments

…into BatchSizeManager

denisgrenader

A few more minor comments, but then I think we are ready to merge once dev testing looks good.

denisgrenader · 2017-05-04T05:26:28Z

src/main/scala/com/box/castle/core/committer/manager/BatchSizeManager.scala

+      if (samples.isEmpty)
+        samples.enqueue(ReadSample(bytesRead, timestamp))
+      else
+        samples.enqueue(ReadSample(samples.last.bytesRead + bytesRead, timestamp))


You can replace this if statement with:
val sample = ReadSample(samples.lastOption.getOrElse(zeroByteSample).bytesRead + bytesRead, timestamp)
samples.enqueue(sample)

in the constructor declare: val zeroByteSample = ReadSample(0, 0)

denisgrenader · 2017-05-04T05:30:02Z

src/main/scala/com/box/castle/core/committer/manager/BatchSizeManager.scala

+      val discount = Math.pow(committerConfig.discountFactor, consecutiveFullBuffers + 1)
+      val computedDelay = Math.min(committerConfig.maxWaitTime.getMillis, (delay * discount).toLong)
+
+      log.info(s"$committerActorId: Idling for ${computedDelay / 1000} seconds after calculating a rate of ${"%.3f".format(rate)} KiB/sec with discountFactor of ${"%.3f".format(discount)}")


This rate is a bit wonky since the original is actually bytes/millisecond which is base 10 not base 2. Let's show it as MiB:
log.info(s"$committerActorId delaying next fetch by ${computedDelay / 1000} seconds after calculating a rate of ${"%.3f".format(rate / 1048.576)} MiB/sec with discountFactor of ${"%.3f".format(discount)}")

denisgrenader · 2017-05-04T05:35:19Z

src/main/scala/com/box/castle/core/committer/states/CommittingBatch.scala

+        if(delay.getMillis > 0)
+          becomeIdling(OffsetAndMetadata(batch.nextOffset, metadata), delay)
+        else
+          becomeFetchingData(OffsetAndMetadata(batch.nextOffset, metadata))


let's add a log line here
log.info("$committerActorId fetching immediately because BatchSizeManager returned a delay of 0")

denisgrenader · 2017-05-04T05:42:27Z

src/main/scala/com/box/castle/core/config/CommitterConfig.scala

@@ -75,4 +85,10 @@ object CommitterConfig {
  val DefaultParallelismFactorByTopic = Map.empty[String, Int]
  val DefaultCorruptMessagePolicy = CorruptMessagePolicy.skip
  val DefaultUseKafkaMetadataManager = true
+  val DefaultUseBatchSizeManager = false
+  val DefaultSamplingSlots = 20
+  val DefaultSamplingInterval = new Duration(60000) // 1 second


the comment here should be // 60 seconds

denisgrenader · 2017-05-04T05:58:23Z

src/main/scala/com/box/castle/core/committer/states/CommittingBatch.scala

+    batchSizeManagerOption match {
+      case Some(batchSizeManager) =>
+        val delay = batchSizeManager.getDelay(committerActorId)
+        if(delay.getMillis > 0)


does joda time have a typed 0 constant? Can this be changed to something like
if (delay > Duration.ZERO) ?

Yes there is but it does not implement comparable I think so '>' doesn't work.

shubhrojroy · 2017-05-05T20:29:24Z

Added BatchSizeManager component to Castle core to control size of batches read from Kafka for a specific committer. Add parameters to control this to Committer config as well.

Adding BatchSizeManager to Castle

03349b5

shubhrojroy requested a review from denisgrenader May 3, 2017 01:27

shubhrojroy assigned shubhrojroy and unassigned shubhrojroy May 3, 2017

denisgrenader requested changes May 3, 2017

View reviewed changes

Changes to BatchSizeManager to handle comments

590ba8d

Shubhro Roy added 2 commits May 3, 2017 19:24

Changes to BatchSizeManager to handle comments

b8196bb

Merge branch 'BatchSizeManager' of https://github.com/Box-Castle/core …

5f72225

…into BatchSizeManager

denisgrenader requested changes May 4, 2017

View reviewed changes

Shubhro Roy added 2 commits May 4, 2017 16:24

Adding targetBatchSizePercent parameter

4cae96d

Addressing Comments

afaf81b

denisgrenader approved these changes May 5, 2017

View reviewed changes

shubhrojroy merged commit a999b43 into master May 5, 2017

shubhrojroy deleted the BatchSizeManager branch May 5, 2017 20:32

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding BatchSizeManager to Castle #3

Adding BatchSizeManager to Castle #3

shubhrojroy commented May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

denisgrenader May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

denisgrenader May 3, 2017

denisgrenader May 3, 2017

shubhrojroy May 3, 2017

shubhrojroy commented May 3, 2017

denisgrenader left a comment

denisgrenader May 4, 2017

shubhrojroy May 4, 2017

denisgrenader May 4, 2017

shubhrojroy May 4, 2017

denisgrenader May 4, 2017

shubhrojroy May 4, 2017

denisgrenader May 4, 2017

denisgrenader May 4, 2017

shubhrojroy May 4, 2017

shubhrojroy commented May 5, 2017

Adding BatchSizeManager to Castle #3

Adding BatchSizeManager to Castle #3

Conversation

shubhrojroy commented May 3, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shubhrojroy commented May 3, 2017

denisgrenader left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shubhrojroy commented May 5, 2017