{"action":"created","issue":{"url":"https://api.github.com/repos/akka/reactive-kafka/issues/110","repository_url":"https://api.github.com/repos/akka/reactive-kafka","labels_url":"https://api.github.com/repos/akka/reactive-kafka/issues/110/labels{/name}","comments_url":"https://api.github.com/repos/akka/reactive-kafka/issues/110/comments","events_url":"https://api.github.com/repos/akka/reactive-kafka/issues/110/events","html_url":"https://github.com/akka/reactive-kafka/issues/110","id":140508089,"number":110,"title":"Apply backpressure per Kafka partition","user":{"login":"atamborrino","id":529577,"avatar_url":"https://avatars.githubusercontent.com/u/529577?v=3","gravatar_id":"","url":"https://api.github.com/users/atamborrino","html_url":"https://github.com/atamborrino","followers_url":"https://api.github.com/users/atamborrino/followers","following_url":"https://api.github.com/users/atamborrino/following{/other_user}","gists_url":"https://api.github.com/users/atamborrino/gists{/gist_id}","starred_url":"https://api.github.com/users/atamborrino/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/atamborrino/subscriptions","organizations_url":"https://api.github.com/users/atamborrino/orgs","repos_url":"https://api.github.com/users/atamborrino/repos","events_url":"https://api.github.com/users/atamborrino/events{/privacy}","received_events_url":"https://api.github.com/users/atamborrino/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":15,"created_at":"2016-03-13T17:24:51Z","updated_at":"2016-04-30T12:43:22Z","closed_at":null,"body":"Hi,\r\n\r\nI open this issue to discuss about a potential improvement.\r\n\r\nFor now, when we create a consumer, back-pressure is applied *per Kafka topic*. If the processing of a record of partition 1 is slow (via a .mapAsync for instance), this will block consumption for all other partitions. This can be mitigated by having several deployed clients with the same consumer-group, but I think we should allow parallel/independent consumption of partitions inside the same instance of client. \r\n\r\nConsuming the partitions of a Kafka topic in parallel in quite standard in the Kafka world, as message ordering is only guaranteed per-partition. Apache Samza and the recent Kafka Streams lib follows this approach by mapping the consumption parallelism level to the number of Kafka partitions (see http://docs.confluent.io/2.1.0-alpha1/streams/architecture.html#parallelism-model). Hence partition processing are totally independent and a slowly processed partition does not block all the others.\r\n\r\nTo do the same with reactive-kafka, a first approach could be the following:\r\n```scala\r\nconsumerSource\r\n  .groupBy(maxNbPartitions, msg => msg.partition)\r\n  .mapAsync(1)(yourAsyncProcessing)\r\n  .mergeSubStreams\r\n  .to(consumer.offsetSink)\r\n```\r\n\r\nHowever, this approach has the following drawbacks:\r\n* Independent processing between partitions is limited: once an input buffer for a slowly processed partition is full, all the other partitions are blocked.\r\n* Non-optimal efficiency: when the processing for a partition begin to be slow, the Kafka consumer will still get in a round-robin fashion records for all partitions, even if partition 1 processing is already late and its input buffer is already big. It would be more efficient to *propagate* this backpressure information to the Kafka client itself to tell it to stop fetching from partition 1 for now and concentrate throughput on other partitions.\r\n\r\nThus, I think that if we want real independent back-pressure between partition we can write a new consumer, say \"fine-grained\", that uses Kafka client 0.9 ConsumerRebalanceListener and Consumption Flow Control (using client.pause(partition) and client.resume(partition), see https://kafka.apache.org/090/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html).\r\n\r\nI see this consumer being a Source[Source[Record]]. The wrapper source being a source of partition, the wrapped source being a source of records. This makes quite sense from a Kafka client point of view (with auto-assigned partitions, a Kafka client is really a source of partition where each partition is a source of records).\r\n\r\nThe code could look like this:\r\n```scala\r\nval perPartitionConsumer = kafka.consumePerPartitionWithOffsetSink(...) // whatever naming\r\nval partitionStream = Source.fromPublisher(perPartitionConsumer.publisher)\r\n\r\npartitionStream.map { recordStream =>\r\n    recordStream\r\n      .mapAsync(1)(yourAsyncProcessing)\r\n}\r\n.flatMapMerge(maxNbPartitions, identity)\r\n.to(perPartitionConsumer.offsetSink)\r\n```\r\n\r\nWith this, we have true per-partition async backpressure. This has also the benefits of exposing the assign/unassign of partitions to the lib users if they want to do something upon these events.\r\n\r\nDo you think it will be interesting to have such fine-grained consumer in the API? If we agree on it, I'll be happy to code it and do a PR.\r\n"},"comment":{"url":"https://api.github.com/repos/akka/reactive-kafka/issues/comments/215961232","html_url":"https://github.com/akka/reactive-kafka/issues/110#issuecomment-215961232","issue_url":"https://api.github.com/repos/akka/reactive-kafka/issues/110","id":215961232,"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},"created_at":"2016-04-30T12:43:21Z","updated_at":"2016-04-30T12:43:21Z","body":"There is discussion about this in PR - https://github.com/akka/reactive-kafka/pull/139"}}