groupWithin causes double execution

See the example below:

package example

import java.util.concurrent.atomic.AtomicLong

import cats.effect._
import cats.syntax.flatMap._
import cats.syntax.functor._
import fs2._

import scala.compat.Platform.ConcurrentModificationException
import scala.concurrent.duration._
import scala.concurrent.blocking

class MiniKafka {
  final val NO_CURRENT_THREAD = -1L
  val currentThread = new AtomicLong(NO_CURRENT_THREAD)

  def acquire() = {
    val threadId = Thread.currentThread().getId
    if (threadId != currentThread.get && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
      throw new ConcurrentModificationException
  }

  def release() = {
    currentThread.set(NO_CURRENT_THREAD)
  }

  def subscribe(): this.type = {
    acquire()
    try this finally release()
  }

  def commit() = {
    acquire()

    try {
    } finally release()
  }

  def close() = {
    acquire()
    try Thread.`yield`()
    finally release()
  }


  def poll(): String = {
    acquire()
    try {
      "abc"
    } finally {
      release()
    }
  }
}

object App extends App {

  def repro[F[_]: ContextShift: Timer](implicit F: Concurrent[F]) = {
    Stream.resource {
      Resource.make(F.delay(new MiniKafka().subscribe()))(c => F.delay(c.close()))
    }.flatMap {
      kafka =>
        Stream.repeatEval {
            F.delay(blocking(kafka.poll())).map(kafka -> _)
        }
    }
      .groupWithin(1000, 1.millis).map(_.toVector)
      .evalMap { x =>
        val kafka = x.unzip._1.head

        F.delay(blocking(kafka.commit())) >>
          F.delay(println(x.unzip._2.toString())).as(kafka)
      }
      .compile.drain
  }

  new IOApp {
    override def run(args: List[String]): IO[ExitCode] =
      repro[cats.effect.IO].as(ExitCode.Success)
  }.main(Array())
}

Run it and leave it for some time, up to ~10 minutes, with current fs2-1.0.2 and JDK11 after some time I get

Vector(abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc, abc)
java.util.ConcurrentModificationException
	at example.MiniKafka.acquire(Repro.scala:22)
	at example.MiniKafka.poll(Repro.scala:49)
	at example.App$.$anonfun$repro$6(Repro.scala:69)
	at scala.concurrent.impl.ExecutionContextImpl$DefaultThreadFactory$$anon$1$$anon$2.block(ExecutionContextImpl.scala:75)
	at java.base/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3118)
	at scala.concurrent.impl.ExecutionContextImpl$DefaultThreadFactory$$anon$1.blockOn(ExecutionContextImpl.scala:87)
	at scala.concurrent.package$.blocking(package.scala:146)
	at example.App$.$anonfun$repro$5(Repro.scala:69)
	at cats.effect.internals.IORunLoop$.cats$effect$internals$IORunLoop$$loop(IORunLoop.scala:87)
	at cats.effect.internals.IORunLoop$RestartCallback.signal(IORunLoop.scala:351)
	at cats.effect.internals.IORunLoop$RestartCallback.apply(IORunLoop.scala:372)
	at cats.effect.internals.IORunLoop$RestartCallback.apply(IORunLoop.scala:312)
	at cats.effect.internals.IOShift$Tick.run(IOShift.scala:36)
	at java.base/java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1426)
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:177)

Which means that poll() was executed twice - in different threads (new thread from blocking). Removing groupWithin or replacing it chunkN or map(Vector(_)) makes this go away.

The same thing happens when running fs2 under scalaz ZIO, so this shouldn't be cats-effect specific.

Looking at groupWithin source I have no idea how this can happen, thoughts?

Start with the groupWithin implementation and compare its behavior with chunkN and map(Vector(_)) using the supplied MiniKafka reproduction. Run the example with fs2-1.0.2 on JDK11 and trace poll execution across threads. Done means groupWithin no longer causes duplicate poll execution or the ConcurrentModificationException.