title: BillingClient.BillingClientStateListener.onBillingSetupFinished is called multiple times
body: <p>I tried to write a wrapper for BillingClient v.2.2.0 with Kotlin Coroutines:</p>
  
  <pre><code>package com.cantalk.photopose.billing
  
  import android.app.Activity
  import android.content.Context
  import com.android.billingclient.api.*
  import com.android.billingclient.api.BillingClient.*
  import com.cantalk.photopose.util.Logger
  import kotlinx.coroutines.CompletableDeferred
  import kotlin.coroutines.resume
  import kotlin.coroutines.resumeWithException
  import kotlin.coroutines.suspendCoroutine
  
  class BillingClientAsync(context: Context) {
      private val billingClient: BillingClient = setupBillingClient(context)
      private val pendingPurchaseFlows = HashMap&lt;String, CompletableDeferred&lt;Purchase&gt;&gt;()
  
      private fun setupBillingClient(context: Context): BillingClient {
          return newBuilder(context)
              .enablePendingPurchases()
              .setListener { billingResult, purchases -&gt;
                  if (billingResult.responseCode == BillingResponseCode.OK &amp;&amp; purchases != null) {
                      for (purchase in purchases) {
                          val deferred = pendingPurchaseFlows.remove(purchase.sku)
                          deferred?.complete(purchase)
                      }
                  } else {
                      val iterator = pendingPurchaseFlows.iterator()
                      while (iterator.hasNext()) {
                          val entry = iterator.next()
                          entry.value.completeExceptionally(BillingException(billingResult))
                          iterator.remove()
                      }
                  }
              }
              .build()
      }
  
      suspend fun queryPurchases(): List&lt;Purchase&gt; {
          Logger.debug("query purchases")
  
          ensureConnected()
          val queryPurchases = billingClient.queryPurchases(SkuType.INAPP)
          if (queryPurchases.responseCode == BillingResponseCode.OK) {
              return queryPurchases.purchasesList
          } else {
              throw BillingException(queryPurchases.billingResult)
          }
      }
  
      suspend fun querySkuDetails(@SkuType type: String, skus: List&lt;String&gt;): List&lt;SkuDetails&gt; {
          Logger.debug("query sku details for", type)
  
          ensureConnected()
          return suspendCoroutine { continuation -&gt;
              val params = SkuDetailsParams.newBuilder()
                  .setType(type)
                  .setSkusList(skus)
                  .build()
              billingClient.querySkuDetailsAsync(params) { billingResult, skuDetailsList -&gt;
                  if (billingResult.responseCode == BillingResponseCode.OK) {
                      continuation.resume(skuDetailsList)
                  } else {
                      continuation.resumeWithException(BillingException(billingResult))
                  }
              }
          }
      }
  
      suspend fun purchase(activity: Activity, skuDetails: SkuDetails): Purchase {
          Logger.debug("purchase", skuDetails.sku)
  
          ensureConnected()
          val currentPurchaseFlow = CompletableDeferred&lt;Purchase&gt;()
              .also { pendingPurchaseFlows[skuDetails.sku] = it }
          val params = BillingFlowParams.newBuilder()
              .setSkuDetails(skuDetails)
              .build()
          billingClient.launchBillingFlow(activity, params)
          return currentPurchaseFlow.await()
      }
  
      suspend fun consume(purchase: Purchase): String {
          Logger.debug("consume", purchase.sku)
  
          ensureConnected()
          return suspendCoroutine { continuation -&gt;
              val params = ConsumeParams.newBuilder()
                  .setPurchaseToken(purchase.purchaseToken)
                  .setDeveloperPayload("TBD")
                  .build()
              billingClient.consumeAsync(params) { billingResult, purchaseToken -&gt;
                  if (billingResult.responseCode == BillingResponseCode.OK) {
                      continuation.resume(purchaseToken)
                  } else {
                      continuation.resumeWithException(BillingException(billingResult))
                  }
              }
          }
      }
  
      suspend fun acknowledgePurchase(purchase: Purchase) {
          Logger.debug("acknowledge", purchase.sku)
  
          ensureConnected()
          return suspendCoroutine { continuation -&gt;
              val params = AcknowledgePurchaseParams.newBuilder()
                  .setPurchaseToken(purchase.purchaseToken)
                  .setDeveloperPayload("TBD")
                  .build()
              billingClient.acknowledgePurchase(params) { billingResult -&gt;
                  if (billingResult.responseCode == BillingResponseCode.OK) {
                      continuation.resume(Unit)
                  } else {
                      continuation.resumeWithException(BillingException(billingResult))
                  }
              }
          }
      }
  
      private suspend fun ensureConnected() {
          if (!billingClient.isReady) {
              startConnection()
          }
      }
  
      private suspend fun startConnection() {
          Logger.debug("connect to billing service")
  
          return suspendCoroutine { continuation -&gt;
              billingClient.startConnection(object : BillingClientStateListener {
                  override fun onBillingSetupFinished(billingResult: BillingResult) {
                      if (billingResult.responseCode == BillingResponseCode.OK) {
                          continuation.resume(Unit)
                      } else {
                          // TODO: 3 Google Play In-app Billing API version is less than 3
                          continuation.resumeWithException(BillingException(billingResult))
                      }
                  }
  
                  override fun onBillingServiceDisconnected() = Unit
              })
          }
      }
  }
  </code></pre>
  
  <p>As you can see, when I try to query purchases or purchase I ensure that client is ready. But in production there are many errors:</p>
  
  <pre><code>java.lang.IllegalStateException: 
    at kotlin.coroutines.SafeContinuation.resumeWith (SafeContinuation.java:2)
    at com.cantalk.photopose.billing.BillingClientAsync$startConnection$2$1.onBillingSetupFinished (BillingClientAsync.java:2)
    at com.android.billingclient.api.zzai.run (zzai.java:6)
  </code></pre>
  
  <p>I tried to understand what the cause of problem and got that if <code>BillingClientStateListener.onBillingSetupFinished</code> will be called multiple time there can be an exception <code>IllegalStateException: Already resumed</code>. I've wondered how is it possible, because I am creating new listener every <code>startConnection</code> call? I can't reproduce this issue on emulator or my test device. Can anybody explain me what does happen here and how to fix it?</p>