Poll heartbeat rate

.github/workflows/build.yml

@@ -0,0 +1,45 @@
+name: build
+
+on: [push, pull_request]
+
+jobs:
+
+  tests:
+    name: scala-${{ matrix.scala }} jdk-${{ matrix.java }} tests
+    runs-on: ubuntu-latest
+
+    strategy:
+      fail-fast: true
+      matrix:
+        java: [8]
+        scala: [2.11.12, 2.12.10]
+
+    steps:
+      - uses: actions/checkout@v2
+      - uses: olafurpg/setup-scala@v10
+        with:
+          java-version: "adopt@1.${{ matrix.java }}"
+
+      - name: Cache SBT Coursier directory
+        uses: actions/cache@v1
+        with:
+          path: ~/.cache/coursier/v1
+          key: ${{ runner.os }}-coursier-${{ hashFiles('**/*.sbt') }}
+          restore-keys: |
+            ${{ runner.os }}-coursier-
+      - name: Cache SBT directory
+        uses: actions/cache@v1
+        with:
+          path: ~/.sbt
+          key: |
+            ${{ runner.os }}-sbt-${{ hashFiles('project/build.properties') }}-${{ hashFiles('project/plugins.sbt') }}
+          restore-keys: ${{ runner.os }}-sbt-
+
+      - name: Run Tests for Kafka 0.10.x, Java ${{ matrix.java }} and Scala ${{ matrix.scala }}
+        run: sbt -J-Xmx6144m  kafka10/test
+
+      - name: Run Tests for Kafka 0.11.x, Java ${{ matrix.java }} and Scala ${{ matrix.scala }}
+        run: sbt -J-Xmx6144m kafka11/test
+
+      - name: Run Tests for Kafka 1.x.x, Java ${{ matrix.java }} and Scala ${{ matrix.scala }}
+        run: sbt -J-Xmx6144m kafka1x/test

.gitignore

@@ -16,3 +16,7 @@ project/plugins/project/
 .scala_dependencies
 .worksheet
 .idea
+
+.bsp/sbt.json
+
+.DS_Store

CHANGES.md

@@ -1,3 +1,22 @@
+
+## Version 1.0.0-RC6 (January 23, 2021)
+
+Depends on Monix 3.3.0
+
+Main changes:
+
+- PR #248: Enable '-Xfatal-warnings' flag (internal)
+- PR #245: Use Kafka Producer Consumer interfaces
+
+## Version 1.0.0-RC7 (April 18, 2020)
+
+Depends on Monix 3.3.0
+
+Main changes:
+
+- PR #248: Enable '-Xfatal-warnings flag
+- PR #245: Use Kafka Producer Consumer interfaces
+
 ## Version 1.0.0-RC6 (April 18, 2020)
 
 Depends on Monix 3.x

README.md

@@ -256,20 +256,18 @@ val observable =
 
 Enjoy!
 
-### Caveats
+### Internal poll heartbeat
 
-[Issue#101](https://github.com/monix/monix-kafka/issues/101)
-Starting from Kafka 0.10.1.0, there is `max.poll.interval.ms` setting:
-
-    The maximum delay between invocations of poll() when using consumer group management. 
-    This places an upper bound on the amount of time that the consumer can be idle before 
-    fetching more records. If poll() is not called before expiration of this timeout, 
-    then the consumer is considered failed and  the group will rebalance in order 
+Starting from Kafka _0.10.1.0_, there is `max.poll.interval.ms` setting that defines the maximum delay between
+invocations of poll(), if it is not called in that interval, then the consumer is considered failed and  the group will rebalance in order 
     to reassign the partitions to another member.
 
-Since, monix-kafka backpressures until all records has been processed. 
-This could be a problem if processing takes time.
-You can reduce `max.poll.records` if you are experiencing this issue.
+This was an [issue](https://github.com/monix/monix-kafka/issues/101) in `monix-kafka`, 
+since poll is not called until all previous consumed ones were processed, so that slow downstream subscribers
+ were in risk of being kicked off the consumer group indefinitely.
+
+It is resolved in `1.0.0-RC8` by introducing an internal poll heartbeat interval
+that runs in the background keeping the consumer alive. 
 
 ## How can I contribute to Monix-Kafka?
 

benchmarks/docker-compose.yml

@@ -0,0 +1,39 @@
+---
+version: '2'
+services:
+
+  zookeeper:
+    image: confluentinc/cp-zookeeper:4.1.1
+    hostname: zookeeper
+    container_name: zookeeper
+    ports:
+      - "2181:2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  broker:
+    image: confluentinc/cp-server:5.5.0
+    hostname: broker
+    container_name: broker
+    depends_on:
+      - zookeeper
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
+      CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181
+      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
+      CONFLUENT_METRICS_ENABLE: 'true'
+      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
+

benchmarks/project/build.properties

@@ -0,0 +1 @@
+sbt.version=1.3.3

benchmarks/results/consumer.md

@@ -0,0 +1,47 @@
+
+## Consumer Benchmarks
+
+The consumer benchmark covers the *manual* and *auto commit* consumer implementations of the different libraries.
+The manual commit will also cover producing committing back the consumed offsets.
+It also runs on different range `pollHeartbeatRate` [1, 10, 15, 100, 1000], which is an important configuration
+ implemented in this `monix-kafka` library.
+
+## RC7
+### 1fork 1thread
+Benchmark                               Mode  Cnt   Score   Error  Units
+ConsumerBenchmark.monix_auto_commit_async  thrpt    7  13.097 ± 0.827  ops/s
+ConsumerBenchmark.monix_auto_commit_sync   thrpt    7  12.486 ± 1.087  ops/s
+ConsumerBenchmark.monix_manual_commit  thrpt   10  11.519 ± 2.247  ops/s
+
+### 1 fork 3 threads
+Benchmark                               Mode  Cnt   Score   Error  Units
+ConsumerBenchmark.monix_auto_commit    thrpt   10  16.186 ± 0.920  ops/s
+ConsumerBenchmark.monix_manual_commit  thrpt   10  16.319 ± 1.465  ops/s
+
+## RC8 - (Introduces PollHeartbeatRate)
+### 1fork 1thread
+---
+Benchmark                                               Mode  Cnt   Score   Error  Units
+ConsumerBenchmark.monixAsyncAutoCommitHeartbeat15ms  thrpt   7  13.126 ± 1.737  ops/s
+ConsumerBenchmark.monixSyncAutoCommitHeartbeat15ms   thrpt   7  12.102 ± 1.214  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat1000ms  thrpt    7   2.978 ± 0.006  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat100ms   thrpt    7   9.961 ± 1.317  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat10ms    thrpt    7  13.346 ± 0.716  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat15ms    thrpt    7  13.454 ± 2.680  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat1ms     thrpt    7  14.281 ± 1.591  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat50ms    thrpt    7  11.900 ± 0.698  ops/s
+---
+### 1 fork 3 threads 
+Benchmark                                               Mode  Cnt   Score   Error  Units
+ConsumerBenchmark.monixAsyncAutoCommitHeartbeat15ms  thrpt   7  16.966 ± 2.659  ops/s
+ConsumerBenchmark.monixSyncAutoCommitHeartbeat15ms   thrpt   7  15.083 ± 4.242  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat1000ms  thrpt    7   2.978 ± 0.006  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat100ms   thrpt    7   9.961 ± 1.317  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat10ms    thrpt    7  13.346 ± 0.716  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat15ms    thrpt    7  13.454 ± 2.680  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat1ms     thrpt    7  14.281 ± 1.591  ops/s
+ConsumerBenchmark.monixManualCommitHeartbeat50ms    thrpt    7  11.900 ± 0.698  ops/s
+
+```sbt
+sbt 'benchmarks/jmh:run -i 5 -wi 1 -f1 -t1 monix.kafka.benchmarks.ConsumerBenchmark.*'
+```

benchmarks/results/producer.md

@@ -0,0 +1,17 @@
+## Producer benchmarks
+
+This section includes benchmarks for single and sink producers.
+
+## RC7
+
+### 10iterations 1fork 1thread
+Benchmark                                 Mode  Cnt  Score   Error  Units
+ProducerBenchmark.monix_single_producer  thrpt    3  0.504 ± 0.668  ops/s
+ProducerBenchmark.monix_sink_producer    thrpt    3  1.861 ± 5.475  ops/s
+
+## RC8 
+
+### 10iterations 1fork 3threads 
+Benchmark                                 Mode  Cnt  Score   Error  Units
+ProducerBenchmark.monix_single_producer  thrpt   10  0.981 ± 0.202  ops/s
+ProducerBenchmark.monix_sink_producer    thrpt   10  3.241 ± 0.191  ops/s

benchmarks/src/main/resources/logback.xml

@@ -0,0 +1,25 @@
+<configuration>
+
+    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+        <layout class="ch.qos.logback.classic.PatternLayout">
+            <Pattern>
+                %d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n
+            </Pattern>
+        </layout>
+    </appender>
+
+    <logger name="monix.kafka" level="info" additivity="false">
+        <appender-ref ref="CONSOLE"/>
+    </logger>
+    <logger name="org.apache.kafka" level="error" additivity="false">
+        <appender-ref ref="CONSOLE"/>
+    </logger>
+    <logger name="org.apache.zookeeper" level="error" additivity="false">
+        <appender-ref ref="CONSOLE"/>
+    </logger>
+
+    <root level="warn">
+        <appender-ref ref="CONSOLE"/>
+    </root>
+
+</configuration>

benchmarks/src/main/scala/monix/kafka/benchmarks/BaseFixture.scala

@@ -0,0 +1,11 @@
+package monix.kafka.benchmarks
+
+import monix.eval.Coeval
+
+import scala.util.Random
+
+trait BaseFixture {
+  protected val brokerUrl = "127.0.0.1:9092"
+  protected val randomId: Coeval[String] = Coeval(Random.alphanumeric.filter(_.isLetter).take(20).mkString)
+  protected val monixTopic = "monix_topic"
+}