Introducing empty `onCompletion` operator with `callbackFlow` changes result

[SanjayVas]

I've got some code using callbackFlow with buffer(Channel.RENDEZVOUS). There's a deadlock elsewhere in my code that I believe is triggered by something in my awaitClose block not completing. Introducing an onCompletion operator, even an empty one, before the buffer call avoids the deadlock. Not having it, or introducing it after buffer, results in the deadlock.

If I'm understanding how the operators are supposed to work, I don't think the presence/placement of an empty onCompletion operator should affect the result.

This is on Kotlin compiler 1.6.21 with kotlinx.coroutines.core 1.6.2 and language level 1.5.

Code sample:

    callbackFlow<Struct> {
      // Some callback code that calls send/close.

      awaitClose {
        if (!coroutineContext.isActive) {
          // Cancel the operation that calls the callback.
        }

        // Clean up the object that calls the callback.
      }
    }
    .onCompletion {
      // For some reason, having an onCompletion specified here is significant.
    }
    .buffer(Channel.RENDEZVOUS)

[SanjayVas]

Full code in case it's relevant: https://github.com/world-federation-of-advertisers/common-jvm/blob/b7fdf217db14e04f3bee2c03aa8645c5c651c498/src/main/kotlin/org/wfanet/measurement/gcloud/spanner/AsyncDatabaseClient.kt#L303

I have an integration test that so far times out deterministically without the onCompletion and passes with it.

[qwwdfsad]

Thanks!

FTR: it is the same issue with under-designed flow fusing which is an internal optimization