{"action":"created","comment":{"url":"https://api.github.com/repos/akka/reactive-kafka/pulls/comments/68017235","id":68017235,"diff_hunk":"@@ -0,0 +1,205 @@\n+/*\n+ * Copyright (C) 2014 - 2016 Softwaremill <http://softwaremill.com>\n+ * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>\n+ */\n+package akka.kafka\n+\n+import java.util\n+import java.util.regex.Pattern\n+\n+import akka.Done\n+import akka.actor.{Actor, ActorRef, Cancellable, Props, Status}\n+import akka.event.LoggingReceive\n+import org.apache.kafka.clients.consumer._\n+import org.apache.kafka.common.TopicPartition\n+\n+import scala.collection.JavaConverters._\n+import scala.concurrent.Promise\n+\n+object KafkaConsumerActor {\n+  case class StoppingException() extends Exception(\"Kafka consumer is stopping\")\n+  def props[K, V](settings: ConsumerSettings[K, V]) = Props(new KafkaConsumerActor(settings))\n+\n+  //requests\n+  private[kafka] case class Assign(tps: Set[TopicPartition])\n+  private[kafka] case class AssignWithOffset(tps: Map[TopicPartition, Long])\n+  private[kafka] case class Subscribe(topics: Set[String], listener: ConsumerRebalanceListener)\n+  private[kafka] case class SubscribePattern(pattern: String, listener: ConsumerRebalanceListener)\n+  private[kafka] case class RequestMessages(topics: Set[TopicPartition])\n+  private[kafka] case object Watch\n+  private[kafka] case object Stop\n+  private[kafka] case class Commit(offsets: Map[TopicPartition, Long])\n+  //responses\n+  private[kafka] case class Assigned(partition: List[TopicPartition])\n+  private[kafka] case class Revoked(partition: List[TopicPartition])\n+  private[kafka] case class Messages[K, V](messages: Iterator[ConsumerRecord[K, V]])\n+  private[kafka] case class Committed(offsets: Map[TopicPartition, OffsetAndMetadata])\n+  //internal\n+  private case object Poll\n+\n+  private var number = 0\n+  private[kafka] def nextNumber() = {\n+    this.synchronized {\n+      number += 1\n+      number\n+    }\n+  }\n+  private[kafka] def rebalanceListener(onAssign: Iterable[TopicPartition] => Unit, onRevoke: Iterable[TopicPartition] => Unit) = new ConsumerRebalanceListener {\n+    override def onPartitionsAssigned(partitions: util.Collection[TopicPartition]): Unit = {\n+      onAssign(partitions.asScala)\n+    }\n+    override def onPartitionsRevoked(partitions: util.Collection[TopicPartition]): Unit = {\n+      onRevoke(partitions.asScala)\n+    }\n+  }\n+}\n+\n+private[kafka] class KafkaConsumerActor[K, V](settings: ConsumerSettings[K, V]) extends Actor {\n+  import KafkaConsumerActor._\n+\n+  def pollTimeout() = settings.pollTimeout\n+  def pollInterval() = settings.pollInterval\n+\n+  var requests = Map.empty[TopicPartition, ActorRef]\n+  var consumer: KafkaConsumer[K, V] = _\n+  var finished: Promise[Done] = _\n+  var nextScheduledPoll: Option[Cancellable] = None\n+  var pollExpected = false\n+  var commitsInProgress = 0\n+  var stopInProgress = false\n+\n+  def receive: Receive = LoggingReceive {\n+    case Watch => sender() ! finished.future\n+    case Assign(tps) =>\n+      val previousAssigned = consumer.assignment()\n+      consumer.assign((tps.toSeq ++ previousAssigned.asScala).asJava)\n+    case AssignWithOffset(tps) =>\n+      consumer.assign(tps.keys.toSeq.asJava)\n+      tps.foreach {\n+        case (tp, offset) => consumer.seek(tp, offset)\n+      }\n+    case Commit(offsets) =>\n+      val commitMap = offsets.mapValues(new OffsetAndMetadata(_))\n+      val reply = sender()\n+      commitsInProgress += 1\n+      consumer.commitAsync(commitMap.asJava, new OffsetCommitCallback {\n+        override def onComplete(offsets: util.Map[TopicPartition, OffsetAndMetadata], exception: Exception): Unit = {\n+          commitsInProgress -= 1\n+          if (exception != null) reply ! Status.Failure(exception)\n+          else reply ! Committed(offsets.asScala.toMap)\n+        }\n+      })\n+      //right now we can not store commits in consumer - https://issues.apache.org/jira/browse/KAFKA-3412\n+      pollExpected = true\n+      poll()\n+    case Subscribe(topics, listener) =>\n+      consumer.subscribe(topics.toList.asJava, listener)\n+    case SubscribePattern(pattern, listener) =>\n+      consumer.subscribe(Pattern.compile(pattern), listener)\n+    case Poll =>\n+      pollExpected = true\n+      poll()\n+    case RequestMessages(topics) =>\n+      requests ++= topics.map(_ -> sender()).toMap\n+      pollExpected = true\n+      poll()\n+    case Stop =>\n+      if (commitsInProgress == 0) {\n+        context.stop(self)\n+      }\n+      else {\n+        stopInProgress = true\n+        context.become(stopping)\n+      }\n+  }\n+\n+  def stopping: Receive = LoggingReceive {\n+    case Watch => sender() ! finished.future\n+    case Assign(tps) =>\n+    case AssignWithOffset(tps) =>\n+    case Commit(offsets) =>\n+      sender() ! Status.Failure(StoppingException())\n+    case Subscribe(topics, listener) =>\n+    case SubscribePattern(pattern, listener) =>\n+    case Poll =>\n+      pollExpected = true\n+      poll()\n+    case RequestMessages(topics) =>\n+      sender() ! Status.Failure(StoppingException())\n+    case Stop =>\n+  }\n+\n+  override def preStart(): Unit = {\n+    super.preStart()\n+    requests = Map.empty[TopicPartition, ActorRef]\n+    consumer = settings.createKafkaConsumer()\n+    nextScheduledPoll = None\n+    commitsInProgress = 0\n+    pollExpected = false\n+    finished = Promise()\n+    schedulePoll()\n+  }\n+\n+  override def postStop(): Unit = {\n+    nextScheduledPoll.foreach(_.cancel())\n+    consumer.close()\n+    finished.trySuccess(Done)\n+    super.postStop()\n+  }\n+\n+  def poll() = {\n+    if (pollExpected) {\n+      pollExpected = false\n+      nextScheduledPoll.foreach(_.cancel())\n+      nextScheduledPoll = None\n+\n+      //set partitions to fetch\n+      val partitionsToFetch = requests.keys.toSet\n+      consumer.assignment().asScala.foreach { tp =>\n+        if (partitionsToFetch.contains(tp)) consumer.resume(tp)\n+        else consumer.pause(tp)\n+      }\n+\n+      val rawResult = consumer.poll(pollTimeout().toMillis)\n+      if (!rawResult.isEmpty) {\n+        // split tps by reply actor\n+        val replyByTP = requests\n+          .groupBy { case (tp, ref) => ref }\n+          .mapValues(_.keys.toSet)\n+\n+        //send messages to actors\n+        replyByTP.foreach {\n+          case (ref, tps) =>\n+            //gather all messages for ref\n+            val messages = tps.foldLeft[Iterator[ConsumerRecord[K, V]]](Iterator.empty) {\n+              case (acc, tp) =>\n+                val tpMessages = rawResult.records(tp).asScala.iterator\n+                if (acc.isEmpty) tpMessages\n+                else acc ++ tpMessages\n+            }\n+            if (messages.nonEmpty) {\n+              ref ! Messages(messages)","path":"core/src/main/scala/akka/kafka/KafkaConsumerActor.scala","position":181,"original_position":181,"commit_id":"7d743d72039e5acf5c7188efa5b34b13f188f4df","original_commit_id":"7d743d72039e5acf5c7188efa5b34b13f188f4df","user":{"login":"drewhk","id":932285,"avatar_url":"https://avatars.githubusercontent.com/u/932285?v=3","gravatar_id":"","url":"https://api.github.com/users/drewhk","html_url":"https://github.com/drewhk","followers_url":"https://api.github.com/users/drewhk/followers","following_url":"https://api.github.com/users/drewhk/following{/other_user}","gists_url":"https://api.github.com/users/drewhk/gists{/gist_id}","starred_url":"https://api.github.com/users/drewhk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/drewhk/subscriptions","organizations_url":"https://api.github.com/users/drewhk/orgs","repos_url":"https://api.github.com/users/drewhk/repos","events_url":"https://api.github.com/users/drewhk/events{/privacy}","received_events_url":"https://api.github.com/users/drewhk/received_events","type":"User","site_admin":false},"body":"this will send an Iterator?","created_at":"2016-06-22T08:58:20Z","updated_at":"2016-06-22T08:58:20Z","html_url":"https://github.com/akka/reactive-kafka/pull/139#discussion_r68017235","pull_request_url":"https://api.github.com/repos/akka/reactive-kafka/pulls/139","_links":{"self":{"href":"https://api.github.com/repos/akka/reactive-kafka/pulls/comments/68017235"},"html":{"href":"https://github.com/akka/reactive-kafka/pull/139#discussion_r68017235"},"pull_request":{"href":"https://api.github.com/repos/akka/reactive-kafka/pulls/139"}}},"pull_request":{"url":"https://api.github.com/repos/akka/reactive-kafka/pulls/139","id":68323441,"html_url":"https://github.com/akka/reactive-kafka/pull/139","diff_url":"https://github.com/akka/reactive-kafka/pull/139.diff","patch_url":"https://github.com/akka/reactive-kafka/pull/139.patch","issue_url":"https://api.github.com/repos/akka/reactive-kafka/issues/139","number":139,"state":"open","locked":false,"title":"TopicPartition API","user":{"login":"13h3r","id":435085,"avatar_url":"https://avatars.githubusercontent.com/u/435085?v=3","gravatar_id":"","url":"https://api.github.com/users/13h3r","html_url":"https://github.com/13h3r","followers_url":"https://api.github.com/users/13h3r/followers","following_url":"https://api.github.com/users/13h3r/following{/other_user}","gists_url":"https://api.github.com/users/13h3r/gists{/gist_id}","starred_url":"https://api.github.com/users/13h3r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/13h3r/subscriptions","organizations_url":"https://api.github.com/users/13h3r/orgs","repos_url":"https://api.github.com/users/13h3r/repos","events_url":"https://api.github.com/users/13h3r/events{/privacy}","received_events_url":"https://api.github.com/users/13h3r/received_events","type":"User","site_admin":false},"body":"This PR is implementations of `TopicPartitionSource`  - `Source[(TopicPartition, Source[Message])]`.\r\n\r\nThere is some small stuff to do, but I want to your feedback on the implementation.\r\n\r\nOccasionally I reimplement all current consumer sources. I did it because the 0.11.M1 implementation contains several design problems:\r\n- Strong coupling between stage and kafka consumer. With this coupling, it is very hard to reuse one kafka consumer from multiple (sub)sources.\r\n- A complete mess in the `CommittableConsumer`. Patrik tried to extract some reusable stuff into a `GraphStageLogic`, but it is still the mess of lifecycle control, poll scheduling and other stuff.\r\n\r\nMy initial thought was `TopicPartitionSource` implementation on the top of the `GraphStage` as we did with `PlainSource` and `CommittableSource`. After quick prototyping, I figured out that hiding kafka consumer behind the `GraphStageLogic` and accessing it via callbacks producing very much boilerplate. I extract kafka consumer with all scheduling stuff and provide an asynchronous interface via `Future`s, but such interface produces a lot of boilerplate in `GraphStageLogic` too. So I decide to move away from `GraphStage` to `ActorPublisher` to be able to communicate with kafka consumer in an asynchronous way.\r\n\r\nAnother challenge is how to provide commit support on top of `PlainSource` without exposing much from it. Having kafka consumer decoupled from stage allows us to pass it to the `CommittableMessage` and build `CommittableMessage` on to of any `ConsumerMessage`.\r\n\r\nWhat we have now:\r\n- `KafkaConsumerActor` as a low-level abstraction on the top of KafkaConsumer\r\n- `PlainSource`, `CommittableSource` and `TopicPatitionSource` implemented with `KafkaConsumerActor`\r\n- `MessageBuilder` trait with `PlainMessageBuilder` and `CommittableMessageBuilder` implementations.\r\n\r\nFeel free to comment the idea and the implementation.\r\n","created_at":"2016-04-29T04:25:06Z","updated_at":"2016-06-22T08:58:20Z","closed_at":null,"merged_at":null,"merge_commit_sha":"8ef193770cde028544984f94c58b37aec5562374","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/akka/reactive-kafka/pulls/139/commits","review_comments_url":"https://api.github.com/repos/akka/reactive-kafka/pulls/139/comments","review_comment_url":"https://api.github.com/repos/akka/reactive-kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/akka/reactive-kafka/issues/139/comments","statuses_url":"https://api.github.com/repos/akka/reactive-kafka/statuses/7d743d72039e5acf5c7188efa5b34b13f188f4df","head":{"label":"13h3r:partition-api","ref":"partition-api","sha":"7d743d72039e5acf5c7188efa5b34b13f188f4df","user":{"login":"13h3r","id":435085,"avatar_url":"https://avatars.githubusercontent.com/u/435085?v=3","gravatar_id":"","url":"https://api.github.com/users/13h3r","html_url":"https://github.com/13h3r","followers_url":"https://api.github.com/users/13h3r/followers","following_url":"https://api.github.com/users/13h3r/following{/other_user}","gists_url":"https://api.github.com/users/13h3r/gists{/gist_id}","starred_url":"https://api.github.com/users/13h3r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/13h3r/subscriptions","organizations_url":"https://api.github.com/users/13h3r/orgs","repos_url":"https://api.github.com/users/13h3r/repos","events_url":"https://api.github.com/users/13h3r/events{/privacy}","received_events_url":"https://api.github.com/users/13h3r/received_events","type":"User","site_admin":false},"repo":{"id":50833158,"name":"reactive-kafka","full_name":"13h3r/reactive-kafka","owner":{"login":"13h3r","id":435085,"avatar_url":"https://avatars.githubusercontent.com/u/435085?v=3","gravatar_id":"","url":"https://api.github.com/users/13h3r","html_url":"https://github.com/13h3r","followers_url":"https://api.github.com/users/13h3r/followers","following_url":"https://api.github.com/users/13h3r/following{/other_user}","gists_url":"https://api.github.com/users/13h3r/gists{/gist_id}","starred_url":"https://api.github.com/users/13h3r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/13h3r/subscriptions","organizations_url":"https://api.github.com/users/13h3r/orgs","repos_url":"https://api.github.com/users/13h3r/repos","events_url":"https://api.github.com/users/13h3r/events{/privacy}","received_events_url":"https://api.github.com/users/13h3r/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/13h3r/reactive-kafka","description":"Reactive Streams API for Apache Kafka","fork":true,"url":"https://api.github.com/repos/13h3r/reactive-kafka","forks_url":"https://api.github.com/repos/13h3r/reactive-kafka/forks","keys_url":"https://api.github.com/repos/13h3r/reactive-kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/13h3r/reactive-kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/13h3r/reactive-kafka/teams","hooks_url":"https://api.github.com/repos/13h3r/reactive-kafka/hooks","issue_events_url":"https://api.github.com/repos/13h3r/reactive-kafka/issues/events{/number}","events_url":"https://api.github.com/repos/13h3r/reactive-kafka/events","assignees_url":"https://api.github.com/repos/13h3r/reactive-kafka/assignees{/user}","branches_url":"https://api.github.com/repos/13h3r/reactive-kafka/branches{/branch}","tags_url":"https://api.github.com/repos/13h3r/reactive-kafka/tags","blobs_url":"https://api.github.com/repos/13h3r/reactive-kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/13h3r/reactive-kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/13h3r/reactive-kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/13h3r/reactive-kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/13h3r/reactive-kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/13h3r/reactive-kafka/languages","stargazers_url":"https://api.github.com/repos/13h3r/reactive-kafka/stargazers","contributors_url":"https://api.github.com/repos/13h3r/reactive-kafka/contributors","subscribers_url":"https://api.github.com/repos/13h3r/reactive-kafka/subscribers","subscription_url":"https://api.github.com/repos/13h3r/reactive-kafka/subscription","commits_url":"https://api.github.com/repos/13h3r/reactive-kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/13h3r/reactive-kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/13h3r/reactive-kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/13h3r/reactive-kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/13h3r/reactive-kafka/contents/{+path}","compare_url":"https://api.github.com/repos/13h3r/reactive-kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/13h3r/reactive-kafka/merges","archive_url":"https://api.github.com/repos/13h3r/reactive-kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/13h3r/reactive-kafka/downloads","issues_url":"https://api.github.com/repos/13h3r/reactive-kafka/issues{/number}","pulls_url":"https://api.github.com/repos/13h3r/reactive-kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/13h3r/reactive-kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/13h3r/reactive-kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/13h3r/reactive-kafka/labels{/name}","releases_url":"https://api.github.com/repos/13h3r/reactive-kafka/releases{/id}","deployments_url":"https://api.github.com/repos/13h3r/reactive-kafka/deployments","created_at":"2016-02-01T10:45:55Z","updated_at":"2016-02-01T10:45:56Z","pushed_at":"2016-06-17T12:40:45Z","git_url":"git://github.com/13h3r/reactive-kafka.git","ssh_url":"git@github.com:13h3r/reactive-kafka.git","clone_url":"https://github.com/13h3r/reactive-kafka.git","svn_url":"https://github.com/13h3r/reactive-kafka","homepage":null,"size":717,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"akka:master","ref":"master","sha":"137780be5feb644f22801fbdaffedfd8aaa62969","user":{"login":"akka","id":1496237,"avatar_url":"https://avatars.githubusercontent.com/u/1496237?v=3","gravatar_id":"","url":"https://api.github.com/users/akka","html_url":"https://github.com/akka","followers_url":"https://api.github.com/users/akka/followers","following_url":"https://api.github.com/users/akka/following{/other_user}","gists_url":"https://api.github.com/users/akka/gists{/gist_id}","starred_url":"https://api.github.com/users/akka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akka/subscriptions","organizations_url":"https://api.github.com/users/akka/orgs","repos_url":"https://api.github.com/users/akka/repos","events_url":"https://api.github.com/users/akka/events{/privacy}","received_events_url":"https://api.github.com/users/akka/received_events","type":"Organization","site_admin":false},"repo":{"id":28635588,"name":"reactive-kafka","full_name":"akka/reactive-kafka","owner":{"login":"akka","id":1496237,"avatar_url":"https://avatars.githubusercontent.com/u/1496237?v=3","gravatar_id":"","url":"https://api.github.com/users/akka","html_url":"https://github.com/akka","followers_url":"https://api.github.com/users/akka/followers","following_url":"https://api.github.com/users/akka/following{/other_user}","gists_url":"https://api.github.com/users/akka/gists{/gist_id}","starred_url":"https://api.github.com/users/akka/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akka/subscriptions","organizations_url":"https://api.github.com/users/akka/orgs","repos_url":"https://api.github.com/users/akka/repos","events_url":"https://api.github.com/users/akka/events{/privacy}","received_events_url":"https://api.github.com/users/akka/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/akka/reactive-kafka","description":"Reactive Streams API for Apache Kafka","fork":false,"url":"https://api.github.com/repos/akka/reactive-kafka","forks_url":"https://api.github.com/repos/akka/reactive-kafka/forks","keys_url":"https://api.github.com/repos/akka/reactive-kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/akka/reactive-kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/akka/reactive-kafka/teams","hooks_url":"https://api.github.com/repos/akka/reactive-kafka/hooks","issue_events_url":"https://api.github.com/repos/akka/reactive-kafka/issues/events{/number}","events_url":"https://api.github.com/repos/akka/reactive-kafka/events","assignees_url":"https://api.github.com/repos/akka/reactive-kafka/assignees{/user}","branches_url":"https://api.github.com/repos/akka/reactive-kafka/branches{/branch}","tags_url":"https://api.github.com/repos/akka/reactive-kafka/tags","blobs_url":"https://api.github.com/repos/akka/reactive-kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/akka/reactive-kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/akka/reactive-kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/akka/reactive-kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/akka/reactive-kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/akka/reactive-kafka/languages","stargazers_url":"https://api.github.com/repos/akka/reactive-kafka/stargazers","contributors_url":"https://api.github.com/repos/akka/reactive-kafka/contributors","subscribers_url":"https://api.github.com/repos/akka/reactive-kafka/subscribers","subscription_url":"https://api.github.com/repos/akka/reactive-kafka/subscription","commits_url":"https://api.github.com/repos/akka/reactive-kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/akka/reactive-kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/akka/reactive-kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/akka/reactive-kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/akka/reactive-kafka/contents/{+path}","compare_url":"https://api.github.com/repos/akka/reactive-kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/akka/reactive-kafka/merges","archive_url":"https://api.github.com/repos/akka/reactive-kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/akka/reactive-kafka/downloads","issues_url":"https://api.github.com/repos/akka/reactive-kafka/issues{/number}","pulls_url":"https://api.github.com/repos/akka/reactive-kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/akka/reactive-kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/akka/reactive-kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/akka/reactive-kafka/labels{/name}","releases_url":"https://api.github.com/repos/akka/reactive-kafka/releases{/id}","deployments_url":"https://api.github.com/repos/akka/reactive-kafka/deployments","created_at":"2014-12-30T15:30:02Z","updated_at":"2016-06-22T06:37:43Z","pushed_at":"2016-06-18T00:54:53Z","git_url":"git://github.com/akka/reactive-kafka.git","ssh_url":"git@github.com:akka/reactive-kafka.git","clone_url":"https://github.com/akka/reactive-kafka.git","svn_url":"https://github.com/akka/reactive-kafka","homepage":null,"size":779,"stargazers_count":541,"watchers_count":541,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":112,"mirror_url":null,"open_issues_count":20,"forks":112,"open_issues":20,"watchers":541,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/akka/reactive-kafka/pulls/139"},"html":{"href":"https://github.com/akka/reactive-kafka/pull/139"},"issue":{"href":"https://api.github.com/repos/akka/reactive-kafka/issues/139"},"comments":{"href":"https://api.github.com/repos/akka/reactive-kafka/issues/139/comments"},"review_comments":{"href":"https://api.github.com/repos/akka/reactive-kafka/pulls/139/comments"},"review_comment":{"href":"https://api.github.com/repos/akka/reactive-kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/akka/reactive-kafka/pulls/139/commits"},"statuses":{"href":"https://api.github.com/repos/akka/reactive-kafka/statuses/7d743d72039e5acf5c7188efa5b34b13f188f4df"}}}}
{"actor":{"display_login":"drewhk"}}