Filter topics by committer #13

bharathi-ravishanker · 2017-07-08T01:49:56Z

No description provided.

bharathi-ravishanker · 2017-07-08T01:50:15Z

Initial commit without tests @denisgrenader

denisgrenader

Mostly name changes, and one logical comment

denisgrenader · 2017-07-08T20:20:27Z

src/main/scala/com/box/castle/Castle.scala

@@ -44,6 +44,11 @@ class Castle private (val castleConfig: CastleConfig, val metricsLogger: Metrics
    }
  }).toMap

+  val filterMap: Map[String, Filter] = committerFactoryMap map {
+    idCommitterFactory => {


idCommitterFactoryPair => {

denisgrenader · 2017-07-08T20:20:50Z

src/main/scala/com/box/castle/Castle.scala

@@ -44,6 +44,11 @@ class Castle private (val castleConfig: CastleConfig, val metricsLogger: Metrics
    }
  }).toMap

+  val filterMap: Map[String, Filter] = committerFactoryMap map {
+    idCommitterFactory => {
+      idCommitterFactory._1 -> idCommitterFactory._2.createFilter()


idCommitterFactoryPair._1 -> idCommitterFactoryPair._2.createTopicFilter()

denisgrenader · 2017-07-08T20:21:40Z

src/main/scala/com/box/castle/Castle.scala

@@ -44,6 +44,11 @@ class Castle private (val castleConfig: CastleConfig, val metricsLogger: Metrics
    }
  }).toMap

+  val filterMap: Map[String, Filter] = committerFactoryMap map {


topicFilterMap: Map[String, TopicFilter] = committerFactoryMap map {

denisgrenader · 2017-07-08T20:22:05Z

src/main/scala/com/box/castle/Castle.scala

@@ -84,7 +89,7 @@ class Castle private (val castleConfig: CastleConfig, val metricsLogger: Metrics

  val workerActorFactory = new WorkerActorFactory(fetcherActorFactory, workerFactory, castleConfig.committerConfigs, metricsLogger)

-  val taskManager = TaskManager(castleConfig.leaderConfig, castleConfig.committerConfigs)
+  val taskManager = TaskManager(castleConfig.leaderConfig, castleConfig.committerConfigs, filterMap)


, topicFilterMap)

denisgrenader · 2017-07-08T20:22:39Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

+  def apply(
+             leaderConfig: LeaderConfig,
+             committerConfigs: Iterable[CommitterConfig],
+             committerFilterMap: Map[String, Filter]) =


topicFilterMap: Map[String, TopicFilter]) =

denisgrenader · 2017-07-08T20:22:55Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

+class TaskManager(
+                   val leaderConfig: LeaderConfig,
+                   val committerConfigs: Iterable[CommitterConfig],
+                   committerFilterMap: Map[String, Filter]) extends Logging {


topicFilterMap: Map[String, TopicFilter]) extends Logging {

denisgrenader · 2017-07-08T20:25:30Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

+  private def matchCommitterFilter(kafkaTopic: KafkaTopic, committerId: String): Boolean = {
+    val filter = committerFilterMap(committerId)
+    filter.containsTopic(kafkaTopic.name)
+  }


val topicFilter = topicFilterMap(committerId)
topicFilter.match(kafkaTopic.name)

match is a keyword in scala and wont let me use it

denisgrenader · 2017-07-08T20:25:45Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

      committerIds + committerId
    else
      committerIds
  }

+  private def matchCommitterFilter(kafkaTopic: KafkaTopic, committerId: String): Boolean = {


private def matchCommitterTopicFilter

denisgrenader · 2017-07-08T20:36:38Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

+      }
+    }
+    else {
+      committerIds


regex match is more stable and well known than whatever the committer filter does, let's invert their checking:

private def matchTopicsRegex(topicsRegex: Regex, kafkaTopic: KafkaTopic, committerId: String, committerIds: Set[String]): Set[String] = { kafkaTopic.name match { case topicsRegex(_*) => if (matchCommitterTopicFilter(kafkaTopic, committerId)) committerIds + committerId else committerIds case _ => committerIds } }

Added comments in the commit. Merging this for now and open to more comments

denisgrenader · 2017-07-11T23:57:05Z

src/test/scala/com/box/castle/core/leader/TaskManagerTest.scala

+      taskManager.matchTopicsSet(Set("performance", "login"),performanceTopic,"falseCommitter", Set("kafkaCommitter")) shouldEqual Set("kafkaCommitter")
+    }
+  }
+


can you add one more test
where the there is a "split" acceptance committer
on some topics
where it accepts some but not others
i.e. if the topics to subscribe are Foo, Bar, Baz
the topic regex is “.*”
so all match
but then the committer returns “true” for Foo,
but “false” for Bar, Baz
right now this case does not cover the split it doesn’t cover the split case

the same thing goes for TopicSet

dashk · 2017-07-11T23:44:19Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

+  def apply(
+             leaderConfig: LeaderConfig,
+             committerConfigs: Iterable[CommitterConfig],
+             topicFilterMap: Map[String, TopicFilter]) =


Should this always be required? If not, would it make sense to provide a default parameter?

Every commiterFactory has a TopicFilter associated with it. It should be mandatory

dashk · 2017-07-11T23:45:20Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

-    }
+  protected[leader] def matchTopicsRegex(topicsRegex: Regex, kafkaTopic: KafkaTopic, committerId: String, committerIds: Set[String]) = {
+      kafkaTopic.name match {
+        case topicsRegex(_*) => if(matchCommitterTopicFilter(kafkaTopic,committerId)) committerIds + committerId else committerIds


Space between if & open paren ("(")

Since we have 2 conditions, would it make sense to change this to a simple if-statement, rather than using case and then a nested If? (See matchTopicsSet(...) -- It seems the logic is quite similar, except the topicsSet(...) versus topicsRegex(...))

dashk · 2017-07-11T23:47:10Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

-      case topicsRegex(_*) => committerIds + committerId
-      case _ => committerIds
-    }
+  protected[leader] def matchTopicsRegex(topicsRegex: Regex, kafkaTopic: KafkaTopic, committerId: String, committerIds: Set[String]) = {


Need a function signature

What's the difference between committerId & committerIds?

dashk · 2017-07-11T23:58:27Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

      committerIds + committerId
    else
      committerIds
  }

+  protected[leader] def matchCommitterTopicFilter(kafkaTopic: KafkaTopic, committerId: String): Boolean = {


It might be good to make function behavior consistent -- matchTopicsSet(...) returns a Set[String], while matchCommitterTopicFilter returns a Boolean.

Function signature 😄

dashk · 2017-07-12T00:00:10Z

src/test/scala/com/box/castle/core/leader/FalseTopicFilter.scala

+import com.box.castle.committer.api.TopicFilter
+
+/**
+  * Created by bravishanker on 7/10/17.


Update class description

dashk · 2017-07-12T00:00:25Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

  }

-  private def matchTopicsSet(topicsSet: Set[String], kafkaTopic: KafkaTopic, committerId: String, committerIds: Set[String]) = {
-    if (topicsSet.contains(kafkaTopic.name))
+  protected[leader] def matchTopicsSet(topicsSet: Set[String], kafkaTopic: KafkaTopic, committerId: String, committerIds: Set[String]) = {


Add function description.

dashk · 2017-07-12T00:00:48Z

src/main/scala/com/box/castle/core/leader/TaskManager.scala

 }



-class TaskManager(val leaderConfig: LeaderConfig, val committerConfigs: Iterable[CommitterConfig]) extends Logging {
+class TaskManager(


Need class description

Filter topics by committer

dbaa438

denisgrenader previously requested changes Jul 8, 2017

View reviewed changes

bravishanker added 2 commits July 10, 2017 12:55

Change variable names

b20d5ed

Unit tests for filtering out topics by committer

b816514

denisgrenader requested changes Jul 11, 2017

View reviewed changes

dashk reviewed Jul 12, 2017

View reviewed changes

bravishanker added 2 commits July 11, 2017 17:52

Adding split committer test, rename, add documentation

979f271

Adding missing test to match set

7cecab6

denisgrenader approved these changes Jul 12, 2017

View reviewed changes

bharathi-ravishanker merged commit 05a6241 into Box-Castle:master Jul 12, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Filter topics by committer #13

Filter topics by committer #13

bharathi-ravishanker commented Jul 8, 2017

bharathi-ravishanker commented Jul 8, 2017 •

edited

Loading

denisgrenader left a comment

denisgrenader Jul 8, 2017

denisgrenader Jul 8, 2017

denisgrenader Jul 8, 2017

denisgrenader Jul 8, 2017

denisgrenader Jul 8, 2017

denisgrenader Jul 8, 2017

denisgrenader Jul 8, 2017

bharathi-ravishanker Jul 11, 2017

denisgrenader Jul 8, 2017

denisgrenader Jul 8, 2017

denisgrenader Jul 11, 2017

dashk Jul 11, 2017

bharathi-ravishanker Jul 12, 2017

dashk Jul 11, 2017

dashk Jul 11, 2017

dashk Jul 11, 2017 •

edited

Loading

dashk Jul 12, 2017

dashk Jul 12, 2017

dashk Jul 12, 2017

Filter topics by committer #13

Filter topics by committer #13

Conversation

bharathi-ravishanker commented Jul 8, 2017

bharathi-ravishanker commented Jul 8, 2017 • edited Loading

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

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

dashk Jul 11, 2017 • edited Loading

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

bharathi-ravishanker commented Jul 8, 2017 •

edited

Loading

dashk Jul 11, 2017 •

edited

Loading