fix blocking initialization

[eugkhp]

I've got this warning by ce3 blocking detection

[WARNING] A Cats Effect worker thread was detected to be in a blocked state (BLOCKED)                                                    │
│   at org.apache.kafka.common.metrics.JmxReporter.reconfigure(JmxReporter.java:109)                                                       │
│   at org.apache.kafka.common.metrics.JmxReporter.configure(JmxReporter.java:93)                                                          │
│   at org.apache.kafka.clients.CommonClientConfigs.metricsReporters(CommonClientConfigs.java:243)                                         │
│   at org.apache.kafka.clients.CommonClientConfigs.metricsReporters(CommonClientConfigs.java:234)                                         │
│   at org.apache.kafka.clients.consumer.internals.ConsumerUtils.createMetrics(ConsumerUtils.java:124)                                     │
│   at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:696)                                                      │
│   at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:671)                                                      │
│   at fs2.kafka.consumer.MkConsumer$.fs2$kafka$consumer$MkConsumer$$anon$1$$_$apply$$anonfun$1(MkConsumer.scala:38)                       │
│   at cats.effect.IOFiber.runLoop(IOFiber.scala:343)                                                                                      │
│   at cats.effect.IOFiber.execR(IOFiber.scala:1362)                                                                                       │
│   at cats.effect.IOFiber.run(IOFiber.scala:112)                                                                                          │
│   at cats.effect.unsafe.WorkerThread.run(WorkerThread.scala:743)                                                                         │
│ This is very likely to be due to suspending a blocking call in IO via                                                                    │
│ `IO.delay` or `IO.apply`. If this is the case then you should use                                                                        │
│ `IO.blocking` or `IO.interruptible` instead.

Looks like something inside new org.apache.kafka.clients.consumer.KafkaConsumer does network call.

[biochimia]

WithConsumer manages the allocation of a dedicated blocking context for the consumer instance provided by MkConsumer. We should probably update WithConsumer to ensure that the instance is constructed in that context, instead of using another blocking context here.

[eugkhp]

Hmm, not entirely sure. Doesn't WithConsumer create a new blocking context for consumer to use? (unless a custom pool is provided)

    val blocking: Resource[F, Blocking[F]] = settings.customBlockingContext match {
      case None     => Blocking.singleThreaded[F]("fs2-kafka-consumer")
      case Some(ec) => Resource.pure(Blocking.fromExecutionContext(ec))
    }

And here we would be using default ce3 context to initialize the consumer .

[biochimia]

Given that a dedicated blocking context is created to invoke every other consumer method. I think it would make most sense to also invoke the constructor in that context.