diff --git a/packages/in_app_purchase/in_app_purchase_android/CHANGELOG.md b/packages/in_app_purchase/in_app_purchase_android/CHANGELOG.md
index f87c7e9bad9..b4cce00f2d1 100644
--- a/packages/in_app_purchase/in_app_purchase_android/CHANGELOG.md
+++ b/packages/in_app_purchase/in_app_purchase_android/CHANGELOG.md
@@ -1,5 +1,7 @@
-## NEXT
+## 0.2.5
 
+* Fixes the management of `BillingClient` connection by handling `BillingResponse.serviceDisconnected`.
+* Introduces `BillingClientManager`.
 * Updates minimum Flutter version to 3.3.
 
 ## 0.2.4+3
diff --git a/packages/in_app_purchase/in_app_purchase_android/lib/billing_client_wrappers.dart b/packages/in_app_purchase/in_app_purchase_android/lib/billing_client_wrappers.dart
index 1dac19f825b..b49be8fe0fe 100644
--- a/packages/in_app_purchase/in_app_purchase_android/lib/billing_client_wrappers.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/lib/billing_client_wrappers.dart
@@ -2,6 +2,7 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+export 'src/billing_client_wrappers/billing_client_manager.dart';
 export 'src/billing_client_wrappers/billing_client_wrapper.dart';
 export 'src/billing_client_wrappers/purchase_wrapper.dart';
 export 'src/billing_client_wrappers/sku_details_wrapper.dart';
diff --git a/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/billing_client_manager.dart b/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/billing_client_manager.dart
new file mode 100644
index 00000000000..0eca29606a5
--- /dev/null
+++ b/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/billing_client_manager.dart
@@ -0,0 +1,154 @@
+// Copyright 2013 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:flutter/widgets.dart';
+
+import 'billing_client_wrapper.dart';
+import 'purchase_wrapper.dart';
+
+/// Abstraction of result of [BillingClient] operation that includes
+/// a [BillingResponse].
+abstract class HasBillingResponse {
+  /// The status of the operation.
+  abstract final BillingResponse responseCode;
+}
+
+/// Utility class that manages a [BillingClient] connection.
+///
+/// Connection is initialized on creation of [BillingClientManager].
+/// If [BillingClient] sends `onBillingServiceDisconnected` event or any
+/// operation returns [BillingResponse.serviceDisconnected], connection is
+/// re-initialized.
+///
+/// [BillingClient] instance is not exposed directly. It can be accessed via
+/// [runWithClient] and [runWithClientNonRetryable] methods that handle the
+/// connection management.
+///
+/// Consider calling [dispose] after the [BillingClient] is no longer needed.
+class BillingClientManager {
+  /// Creates the [BillingClientManager].
+  ///
+  /// Immediately initializes connection to the underlying [BillingClient].
+  BillingClientManager() {
+    _connect();
+  }
+
+  /// Stream of `onPurchasesUpdated` events from the [BillingClient].
+  ///
+  /// This is a broadcast stream, so it can be listened to multiple times.
+  /// A "done" event will be sent after [dispose] is called.
+  late final Stream<PurchasesResultWrapper> purchasesUpdatedStream =
+      _purchasesUpdatedController.stream;
+
+  /// [BillingClient] instance managed by this [BillingClientManager].
+  ///
+  /// In order to access the [BillingClient], use [runWithClient]
+  /// and [runWithClientNonRetryable] methods.
+  @visibleForTesting
+  late final BillingClient client = BillingClient(_onPurchasesUpdated);
+
+  final StreamController<PurchasesResultWrapper> _purchasesUpdatedController =
+      StreamController<PurchasesResultWrapper>.broadcast();
+
+  bool _isConnecting = false;
+  bool _isDisposed = false;
+
+  // Initialized immediately in the constructor, so it's always safe to access.
+  late Future<void> _readyFuture;
+
+  /// Executes the given [action] with access to the underlying [BillingClient].
+  ///
+  /// If necessary, waits for the underlying [BillingClient] to connect.
+  /// If given [action] returns [BillingResponse.serviceDisconnected], it will
+  /// be transparently retried after the connection is restored. Because
+  /// of this, [action] may be called multiple times.
+  ///
+  /// A response with [BillingResponse.serviceDisconnected] may be returned
+  /// in case of [dispose] being called during the operation.
+  ///
+  /// See [runWithClientNonRetryable] for operations that do not return
+  /// a subclass of [HasBillingResponse].
+  Future<R> runWithClient<R extends HasBillingResponse>(
+    Future<R> Function(BillingClient client) action,
+  ) async {
+    _debugAssertNotDisposed();
+    await _readyFuture;
+    final R result = await action(client);
+    if (result.responseCode == BillingResponse.serviceDisconnected &&
+        !_isDisposed) {
+      await _connect();
+      return runWithClient(action);
+    } else {
+      return result;
+    }
+  }
+
+  /// Executes the given [action] with access to the underlying [BillingClient].
+  ///
+  /// If necessary, waits for the underlying [BillingClient] to connect.
+  /// Designed only for operations that do not return a subclass
+  /// of [HasBillingResponse] (e.g. [BillingClient.isReady],
+  /// [BillingClient.isFeatureSupported]).
+  ///
+  /// See [runWithClient] for operations that return a subclass
+  /// of [HasBillingResponse].
+  Future<R> runWithClientNonRetryable<R>(
+    Future<R> Function(BillingClient client) action,
+  ) async {
+    _debugAssertNotDisposed();
+    await _readyFuture;
+    return action(client);
+  }
+
+  /// Ends connection to the [BillingClient].
+  ///
+  /// Consider calling [dispose] after you no longer need the [BillingClient]
+  /// API to free up the resources.
+  ///
+  /// After calling [dispose]:
+  /// - Further connection attempts will not be made.
+  /// - [purchasesUpdatedStream] will be closed.
+  /// - Calls to [runWithClient] and [runWithClientNonRetryable] will throw.
+  void dispose() {
+    _debugAssertNotDisposed();
+    _isDisposed = true;
+    client.endConnection();
+    _purchasesUpdatedController.close();
+  }
+
+  // If disposed, does nothing.
+  // If currently connecting, waits for it to complete.
+  // Otherwise, starts a new connection.
+  Future<void> _connect() {
+    if (_isDisposed) {
+      return Future<void>.value();
+    }
+    if (_isConnecting) {
+      return _readyFuture;
+    }
+    _isConnecting = true;
+    _readyFuture = Future<void>.sync(() async {
+      await client.startConnection(onBillingServiceDisconnected: _connect);
+      _isConnecting = false;
+    });
+    return _readyFuture;
+  }
+
+  void _onPurchasesUpdated(PurchasesResultWrapper event) {
+    if (_isDisposed) {
+      return;
+    }
+    _purchasesUpdatedController.add(event);
+  }
+
+  void _debugAssertNotDisposed() {
+    assert(
+      !_isDisposed,
+      'A BillingClientManager was used after being disposed. Once you have '
+      'called dispose() on a BillingClientManager, it can no longer be used.',
+    );
+  }
+}
diff --git a/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/billing_client_wrapper.dart b/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/billing_client_wrapper.dart
index 2d4a3f96b50..04a73f6c564 100644
--- a/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/billing_client_wrapper.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/billing_client_wrapper.dart
@@ -51,6 +51,12 @@ typedef PurchasesUpdatedListener = void Function(
 /// `com.android.billingclient.api.BillingClient` API as much as possible, with
 /// some minor changes to account for language differences. Callbacks have been
 /// converted to futures where appropriate.
+///
+/// Connection to [BillingClient] may be lost at any time (see
+/// `onBillingServiceDisconnected` param of [startConnection] and
+/// [BillingResponse.serviceDisconnected]).
+/// Consider using [BillingClientManager] that handles these disconnections
+/// transparently.
 class BillingClient {
   /// Creates a billing client.
   BillingClient(PurchasesUpdatedListener onPurchasesUpdated) {
diff --git a/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/purchase_wrapper.dart b/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/purchase_wrapper.dart
index 4e6b953096e..633aa732165 100644
--- a/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/purchase_wrapper.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/purchase_wrapper.dart
@@ -6,6 +6,7 @@ import 'package:flutter/foundation.dart';
 import 'package:in_app_purchase_platform_interface/in_app_purchase_platform_interface.dart';
 import 'package:json_annotation/json_annotation.dart';
 
+import 'billing_client_manager.dart';
 import 'billing_client_wrapper.dart';
 import 'sku_details_wrapper.dart';
 
@@ -265,7 +266,7 @@ class PurchaseHistoryRecordWrapper {
 @JsonSerializable()
 @BillingResponseConverter()
 @immutable
-class PurchasesResultWrapper {
+class PurchasesResultWrapper implements HasBillingResponse {
   /// Creates a [PurchasesResultWrapper] with the given purchase result details.
   const PurchasesResultWrapper(
       {required this.responseCode,
@@ -300,6 +301,7 @@ class PurchasesResultWrapper {
   ///
   /// This can represent either the status of the "query purchase history" half
   /// of the operation and the "user made purchases" transaction itself.
+  @override
   final BillingResponse responseCode;
 
   /// The list of successful purchases made in this transaction.
@@ -316,7 +318,7 @@ class PurchasesResultWrapper {
 @JsonSerializable()
 @BillingResponseConverter()
 @immutable
-class PurchasesHistoryResult {
+class PurchasesHistoryResult implements HasBillingResponse {
   /// Creates a [PurchasesHistoryResult] with the provided history.
   const PurchasesHistoryResult(
       {required this.billingResult, required this.purchaseHistoryRecordList});
@@ -325,6 +327,9 @@ class PurchasesHistoryResult {
   factory PurchasesHistoryResult.fromJson(Map<String, dynamic> map) =>
       _$PurchasesHistoryResultFromJson(map);
 
+  @override
+  BillingResponse get responseCode => billingResult.responseCode;
+
   @override
   bool operator ==(Object other) {
     if (identical(other, this)) {
diff --git a/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/sku_details_wrapper.dart b/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/sku_details_wrapper.dart
index 1c5c2d1fcee..2689cf37eac 100644
--- a/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/sku_details_wrapper.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/lib/src/billing_client_wrappers/sku_details_wrapper.dart
@@ -5,6 +5,7 @@
 import 'package:flutter/foundation.dart';
 import 'package:json_annotation/json_annotation.dart';
 
+import 'billing_client_manager.dart';
 import 'billing_client_wrapper.dart';
 
 // WARNING: Changes to `@JsonSerializable` classes need to be reflected in the
@@ -182,7 +183,7 @@ class SkuDetailsWrapper {
 /// Returned by [BillingClient.querySkuDetails].
 @JsonSerializable()
 @immutable
-class SkuDetailsResponseWrapper {
+class SkuDetailsResponseWrapper implements HasBillingResponse {
   /// Creates a [SkuDetailsResponseWrapper] with the given purchase details.
   @visibleForTesting
   const SkuDetailsResponseWrapper(
@@ -202,6 +203,9 @@ class SkuDetailsResponseWrapper {
   @JsonKey(defaultValue: <SkuDetailsWrapper>[])
   final List<SkuDetailsWrapper> skuDetailsList;
 
+  @override
+  BillingResponse get responseCode => billingResult.responseCode;
+
   @override
   bool operator ==(Object other) {
     if (other.runtimeType != runtimeType) {
@@ -221,7 +225,7 @@ class SkuDetailsResponseWrapper {
 @JsonSerializable()
 @BillingResponseConverter()
 @immutable
-class BillingResultWrapper {
+class BillingResultWrapper implements HasBillingResponse {
   /// Constructs the object with [responseCode] and [debugMessage].
   const BillingResultWrapper({required this.responseCode, this.debugMessage});
 
@@ -239,6 +243,7 @@ class BillingResultWrapper {
   }
 
   /// Response code returned in the Play Billing API calls.
+  @override
   final BillingResponse responseCode;
 
   /// Debug message returned in the Play Billing API calls.
diff --git a/packages/in_app_purchase/in_app_purchase_android/lib/src/in_app_purchase_android_platform.dart b/packages/in_app_purchase/in_app_purchase_android/lib/src/in_app_purchase_android_platform.dart
index c8046d6e655..b605c2f611c 100644
--- a/packages/in_app_purchase/in_app_purchase_android/lib/src/in_app_purchase_android_platform.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/lib/src/in_app_purchase_android_platform.dart
@@ -29,18 +29,13 @@ const String kIAPSource = 'google_play';
 /// generic plugin API.
 class InAppPurchaseAndroidPlatform extends InAppPurchasePlatform {
   InAppPurchaseAndroidPlatform._() {
-    billingClient = BillingClient((PurchasesResultWrapper resultWrapper) async {
-      _purchaseUpdatedController
-          .add(await _getPurchaseDetailsFromResult(resultWrapper));
-    });
-
     // Register [InAppPurchaseAndroidPlatformAddition].
     InAppPurchasePlatformAddition.instance =
-        InAppPurchaseAndroidPlatformAddition(billingClient);
+        InAppPurchaseAndroidPlatformAddition(billingClientManager);
 
-    _readyFuture = _connect();
-    _purchaseUpdatedController =
-        StreamController<List<PurchaseDetails>>.broadcast();
+    billingClientManager.purchasesUpdatedStream
+        .asyncMap(_getPurchaseDetailsFromResult)
+        .listen(_purchaseUpdatedController.add);
   }
 
   /// Registers this class as the default instance of [InAppPurchasePlatform].
@@ -50,26 +45,25 @@ class InAppPurchaseAndroidPlatform extends InAppPurchasePlatform {
     InAppPurchasePlatform.instance = InAppPurchaseAndroidPlatform._();
   }
 
-  static late StreamController<List<PurchaseDetails>>
-      _purchaseUpdatedController;
+  final StreamController<List<PurchaseDetails>> _purchaseUpdatedController =
+      StreamController<List<PurchaseDetails>>.broadcast();
 
   @override
-  Stream<List<PurchaseDetails>> get purchaseStream =>
+  late final Stream<List<PurchaseDetails>> purchaseStream =
       _purchaseUpdatedController.stream;
 
   /// The [BillingClient] that's abstracted by [GooglePlayConnection].
   ///
   /// This field should not be used out of test code.
   @visibleForTesting
-  late final BillingClient billingClient;
+  final BillingClientManager billingClientManager = BillingClientManager();
 
-  late Future<void> _readyFuture;
   static final Set<String> _productIdsToConsume = <String>{};
 
   @override
   Future<bool> isAvailable() async {
-    await _readyFuture;
-    return billingClient.isReady();
+    return billingClientManager
+        .runWithClientNonRetryable((BillingClient client) => client.isReady());
   }
 
   @override
@@ -77,27 +71,33 @@ class InAppPurchaseAndroidPlatform extends InAppPurchasePlatform {
       Set<String> identifiers) async {
     List<SkuDetailsResponseWrapper> responses;
     PlatformException? exception;
+
+    Future<SkuDetailsResponseWrapper> querySkuDetails(SkuType type) {
+      return billingClientManager.runWithClient(
+        (BillingClient client) => client.querySkuDetails(
+          skuType: type,
+          skusList: identifiers.toList(),
+        ),
+      );
+    }
+
     try {
       responses = await Future.wait(<Future<SkuDetailsResponseWrapper>>[
-        billingClient.querySkuDetails(
-            skuType: SkuType.inapp, skusList: identifiers.toList()),
-        billingClient.querySkuDetails(
-            skuType: SkuType.subs, skusList: identifiers.toList())
+        querySkuDetails(SkuType.inapp),
+        querySkuDetails(SkuType.subs),
       ]);
     } on PlatformException catch (e) {
       exception = e;
-      responses = <SkuDetailsResponseWrapper>[
-        // ignore: invalid_use_of_visible_for_testing_member
-        SkuDetailsResponseWrapper(
-            billingResult: BillingResultWrapper(
-                responseCode: BillingResponse.error, debugMessage: e.code),
-            skuDetailsList: const <SkuDetailsWrapper>[]),
-        // ignore: invalid_use_of_visible_for_testing_member
-        SkuDetailsResponseWrapper(
-            billingResult: BillingResultWrapper(
-                responseCode: BillingResponse.error, debugMessage: e.code),
-            skuDetailsList: const <SkuDetailsWrapper>[])
-      ];
+      // ignore: invalid_use_of_visible_for_testing_member
+      final SkuDetailsResponseWrapper response = SkuDetailsResponseWrapper(
+        billingResult: BillingResultWrapper(
+          responseCode: BillingResponse.error,
+          debugMessage: e.code,
+        ),
+        skuDetailsList: const <SkuDetailsWrapper>[],
+      );
+      // Error response for both queries should be the same, so we can reuse it.
+      responses = <SkuDetailsResponseWrapper>[response, response];
     }
     final List<ProductDetails> productDetailsList =
         responses.expand((SkuDetailsResponseWrapper response) {
@@ -132,13 +132,16 @@ class InAppPurchaseAndroidPlatform extends InAppPurchasePlatform {
     }
 
     final BillingResultWrapper billingResultWrapper =
-        await billingClient.launchBillingFlow(
-            sku: purchaseParam.productDetails.id,
-            accountId: purchaseParam.applicationUserName,
-            oldSku: changeSubscriptionParam?.oldPurchaseDetails.productID,
-            purchaseToken: changeSubscriptionParam
-                ?.oldPurchaseDetails.verificationData.serverVerificationData,
-            prorationMode: changeSubscriptionParam?.prorationMode);
+        await billingClientManager.runWithClient(
+      (BillingClient client) => client.launchBillingFlow(
+        sku: purchaseParam.productDetails.id,
+        accountId: purchaseParam.applicationUserName,
+        oldSku: changeSubscriptionParam?.oldPurchaseDetails.productID,
+        purchaseToken: changeSubscriptionParam
+            ?.oldPurchaseDetails.verificationData.serverVerificationData,
+        prorationMode: changeSubscriptionParam?.prorationMode,
+      ),
+    );
     return billingResultWrapper.responseCode == BillingResponse.ok;
   }
 
@@ -171,8 +174,10 @@ class InAppPurchaseAndroidPlatform extends InAppPurchasePlatform {
           'completePurchase unsuccessful. The `purchase.verificationData` is not valid');
     }
 
-    return billingClient
-        .acknowledgePurchase(purchase.verificationData.serverVerificationData);
+    return billingClientManager.runWithClient(
+      (BillingClient client) => client.acknowledgePurchase(
+          purchase.verificationData.serverVerificationData),
+    );
   }
 
   @override
@@ -182,8 +187,12 @@ class InAppPurchaseAndroidPlatform extends InAppPurchasePlatform {
     List<PurchasesResultWrapper> responses;
 
     responses = await Future.wait(<Future<PurchasesResultWrapper>>[
-      billingClient.queryPurchases(SkuType.inapp),
-      billingClient.queryPurchases(SkuType.subs)
+      billingClientManager.runWithClient(
+        (BillingClient client) => client.queryPurchases(SkuType.inapp),
+      ),
+      billingClientManager.runWithClient(
+        (BillingClient client) => client.queryPurchases(SkuType.subs),
+      ),
     ]);
 
     final Set<String> errorCodeSet = responses
@@ -219,9 +228,6 @@ class InAppPurchaseAndroidPlatform extends InAppPurchasePlatform {
     _purchaseUpdatedController.add(pastPurchases);
   }
 
-  Future<void> _connect() =>
-      billingClient.startConnection(onBillingServiceDisconnected: () {});
-
   Future<PurchaseDetails> _maybeAutoConsumePurchase(
       PurchaseDetails purchaseDetails) async {
     if (!(purchaseDetails.status == PurchaseStatus.purchased &&
@@ -279,15 +285,16 @@ class InAppPurchaseAndroidPlatform extends InAppPurchasePlatform {
       }
       return <PurchaseDetails>[
         PurchaseDetails(
-            purchaseID: '',
-            productID: '',
-            status: status,
-            transactionDate: null,
-            verificationData: PurchaseVerificationData(
-                localVerificationData: '',
-                serverVerificationData: '',
-                source: kIAPSource))
-          ..error = error
+          purchaseID: '',
+          productID: '',
+          status: status,
+          transactionDate: null,
+          verificationData: PurchaseVerificationData(
+            localVerificationData: '',
+            serverVerificationData: '',
+            source: kIAPSource,
+          ),
+        )..error = error
       ];
     }
   }
diff --git a/packages/in_app_purchase/in_app_purchase_android/lib/src/in_app_purchase_android_platform_addition.dart b/packages/in_app_purchase/in_app_purchase_android/lib/src/in_app_purchase_android_platform_addition.dart
index d5657d1a38d..67e21dfad8f 100644
--- a/packages/in_app_purchase/in_app_purchase_android/lib/src/in_app_purchase_android_platform_addition.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/lib/src/in_app_purchase_android_platform_addition.dart
@@ -12,8 +12,8 @@ import '../in_app_purchase_android.dart';
 class InAppPurchaseAndroidPlatformAddition
     extends InAppPurchasePlatformAddition {
   /// Creates a [InAppPurchaseAndroidPlatformAddition] which uses the supplied
-  /// `BillingClient` to provide Android specific features.
-  InAppPurchaseAndroidPlatformAddition(this._billingClient);
+  /// `BillingClientManager` to provide Android specific features.
+  InAppPurchaseAndroidPlatformAddition(this._billingClientManager);
 
   /// Whether pending purchase is enabled.
   ///
@@ -42,7 +42,7 @@ class InAppPurchaseAndroidPlatformAddition
     // No-op, until it is time to completely remove this method from the API.
   }
 
-  final BillingClient _billingClient;
+  final BillingClientManager _billingClientManager;
 
   /// Mark that the user has consumed a product.
   ///
@@ -54,8 +54,10 @@ class InAppPurchaseAndroidPlatformAddition
       throw ArgumentError(
           'consumePurchase unsuccessful. The `purchase.verificationData` is not valid');
     }
-    return _billingClient
-        .consumeAsync(purchase.verificationData.serverVerificationData);
+    return _billingClientManager.runWithClient(
+      (BillingClient client) =>
+          client.consumeAsync(purchase.verificationData.serverVerificationData),
+    );
   }
 
   /// Query all previous purchases.
@@ -78,8 +80,12 @@ class InAppPurchaseAndroidPlatformAddition
     PlatformException? exception;
     try {
       responses = await Future.wait(<Future<PurchasesResultWrapper>>[
-        _billingClient.queryPurchases(SkuType.inapp),
-        _billingClient.queryPurchases(SkuType.subs)
+        _billingClientManager.runWithClient(
+          (BillingClient client) => client.queryPurchases(SkuType.inapp),
+        ),
+        _billingClientManager.runWithClient(
+          (BillingClient client) => client.queryPurchases(SkuType.subs),
+        ),
       ]);
     } on PlatformException catch (e) {
       exception = e;
@@ -141,7 +147,9 @@ class InAppPurchaseAndroidPlatformAddition
   /// Checks if the specified feature or capability is supported by the Play Store.
   /// Call this to check if a [BillingClientFeature] is supported by the device.
   Future<bool> isFeatureSupported(BillingClientFeature feature) async {
-    return _billingClient.isFeatureSupported(feature);
+    return _billingClientManager.runWithClientNonRetryable(
+      (BillingClient client) => client.isFeatureSupported(feature),
+    );
   }
 
   /// Initiates a flow to confirm the change of price for an item subscribed by the user.
@@ -153,6 +161,9 @@ class InAppPurchaseAndroidPlatformAddition
   /// [InAppPurchaseAndroidPlatform.queryProductDetails] call.
   Future<BillingResultWrapper> launchPriceChangeConfirmationFlow(
       {required String sku}) {
-    return _billingClient.launchPriceChangeConfirmationFlow(sku: sku);
+    return _billingClientManager.runWithClient(
+      (BillingClient client) =>
+          client.launchPriceChangeConfirmationFlow(sku: sku),
+    );
   }
 }
diff --git a/packages/in_app_purchase/in_app_purchase_android/pubspec.yaml b/packages/in_app_purchase/in_app_purchase_android/pubspec.yaml
index 1acaba34638..0d9dc31d290 100644
--- a/packages/in_app_purchase/in_app_purchase_android/pubspec.yaml
+++ b/packages/in_app_purchase/in_app_purchase_android/pubspec.yaml
@@ -2,7 +2,7 @@ name: in_app_purchase_android
 description: An implementation for the Android platform of the Flutter `in_app_purchase` plugin. This uses the Android BillingClient APIs.
 repository: https://github.com/flutter/packages/tree/main/packages/in_app_purchase/in_app_purchase_android
 issue_tracker: https://github.com/flutter/flutter/issues?q=is%3Aissue+is%3Aopen+label%3A%22p%3A+in_app_purchase%22
-version: 0.2.4+3
+version: 0.2.5
 
 environment:
   sdk: ">=2.18.0 <4.0.0"
diff --git a/packages/in_app_purchase/in_app_purchase_android/test/billing_client_wrappers/billing_client_manager_test.dart b/packages/in_app_purchase/in_app_purchase_android/test/billing_client_wrappers/billing_client_manager_test.dart
new file mode 100644
index 00000000000..9fd76f4aad6
--- /dev/null
+++ b/packages/in_app_purchase/in_app_purchase_android/test/billing_client_wrappers/billing_client_manager_test.dart
@@ -0,0 +1,107 @@
+// Copyright 2013 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:flutter/services.dart';
+import 'package:flutter/widgets.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:in_app_purchase_android/billing_client_wrappers.dart';
+import 'package:in_app_purchase_android/src/channel.dart';
+
+import '../stub_in_app_purchase_platform.dart';
+import 'purchase_wrapper_test.dart';
+
+void main() {
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  final StubInAppPurchasePlatform stubPlatform = StubInAppPurchasePlatform();
+  late BillingClientManager manager;
+  late Completer<void> connectedCompleter;
+
+  const String startConnectionCall =
+      'BillingClient#startConnection(BillingClientStateListener)';
+  const String endConnectionCall = 'BillingClient#endConnection()';
+  const String onBillingServiceDisconnectedCallback =
+      'BillingClientStateListener#onBillingServiceDisconnected()';
+
+  setUpAll(() =>
+      channel.setMockMethodCallHandler(stubPlatform.fakeMethodCallHandler));
+
+  setUp(() {
+    WidgetsFlutterBinding.ensureInitialized();
+    connectedCompleter = Completer<void>.sync();
+    stubPlatform.addResponse(
+      name: startConnectionCall,
+      value: buildBillingResultMap(
+        const BillingResultWrapper(responseCode: BillingResponse.ok),
+      ),
+      additionalStepBeforeReturn: (dynamic _) => connectedCompleter.future,
+    );
+    stubPlatform.addResponse(name: endConnectionCall);
+    manager = BillingClientManager();
+  });
+
+  tearDown(() => stubPlatform.reset());
+
+  group('BillingClientWrapper', () {
+    test('connects on initialization', () {
+      expect(stubPlatform.countPreviousCalls(startConnectionCall), equals(1));
+    });
+
+    test('waits for connection before executing the operations', () {
+      bool called1 = false;
+      bool called2 = false;
+      manager.runWithClient((BillingClient _) async {
+        called1 = true;
+        return const BillingResultWrapper(responseCode: BillingResponse.ok);
+      });
+      manager.runWithClientNonRetryable(
+        (BillingClient _) async => called2 = true,
+      );
+      expect(called1, equals(false));
+      expect(called2, equals(false));
+      connectedCompleter.complete();
+      expect(called1, equals(true));
+      expect(called2, equals(true));
+    });
+
+    test('re-connects when client sends onBillingServiceDisconnected', () {
+      connectedCompleter.complete();
+      manager.client.callHandler(
+        const MethodCall(onBillingServiceDisconnectedCallback,
+            <String, dynamic>{'handle': 0}),
+      );
+      expect(stubPlatform.countPreviousCalls(startConnectionCall), equals(2));
+    });
+
+    test(
+      're-connects when operation returns BillingResponse.serviceDisconnected',
+      () async {
+        connectedCompleter.complete();
+        int timesCalled = 0;
+        final BillingResultWrapper result = await manager.runWithClient(
+          (BillingClient _) async {
+            timesCalled++;
+            return BillingResultWrapper(
+              responseCode: timesCalled == 1
+                  ? BillingResponse.serviceDisconnected
+                  : BillingResponse.ok,
+            );
+          },
+        );
+        expect(stubPlatform.countPreviousCalls(startConnectionCall), equals(2));
+        expect(timesCalled, equals(2));
+        expect(result.responseCode, equals(BillingResponse.ok));
+      },
+    );
+
+    test('does not re-connect when disposed', () {
+      connectedCompleter.complete();
+      manager.dispose();
+      expect(stubPlatform.countPreviousCalls(startConnectionCall), equals(1));
+      expect(stubPlatform.countPreviousCalls(endConnectionCall), equals(1));
+    });
+  });
+}
diff --git a/packages/in_app_purchase/in_app_purchase_android/test/in_app_purchase_android_platform_addition_test.dart b/packages/in_app_purchase/in_app_purchase_android/test/in_app_purchase_android_platform_addition_test.dart
index a97c69608a3..1b61f53b0d3 100644
--- a/packages/in_app_purchase/in_app_purchase_android/test/in_app_purchase_android_platform_addition_test.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/test/in_app_purchase_android_platform_addition_test.dart
@@ -39,7 +39,7 @@ void main() {
         value: buildBillingResultMap(expectedBillingResult));
     stubPlatform.addResponse(name: endConnectionCall);
     iapAndroidPlatformAddition =
-        InAppPurchaseAndroidPlatformAddition(BillingClient((_) {}));
+        InAppPurchaseAndroidPlatformAddition(BillingClientManager());
   });
 
   group('consume purchases', () {
diff --git a/packages/in_app_purchase/in_app_purchase_android/test/in_app_purchase_android_platform_test.dart b/packages/in_app_purchase/in_app_purchase_android/test/in_app_purchase_android_platform_test.dart
index 347bacde20b..a679def27d5 100644
--- a/packages/in_app_purchase/in_app_purchase_android/test/in_app_purchase_android_platform_test.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/test/in_app_purchase_android_platform_test.dart
@@ -24,6 +24,10 @@ void main() {
   const String startConnectionCall =
       'BillingClient#startConnection(BillingClientStateListener)';
   const String endConnectionCall = 'BillingClient#endConnection()';
+  const String acknowledgePurchaseCall =
+      'BillingClient#(AcknowledgePurchaseParams params, (AcknowledgePurchaseParams, AcknowledgePurchaseResponseListener)';
+  const String onBillingServiceDisconnectedCallback =
+      'BillingClientStateListener#onBillingServiceDisconnected()';
 
   setUpAll(() {
     _ambiguate(TestDefaultBinaryMessengerBinding.instance)!
@@ -57,6 +61,45 @@ void main() {
       //await iapAndroidPlatform.isAvailable();
       expect(stubPlatform.countPreviousCalls(startConnectionCall), equals(1));
     });
+
+    test('re-connects when client sends onBillingServiceDisconnected', () {
+      iapAndroidPlatform.billingClientManager.client.callHandler(
+        const MethodCall(onBillingServiceDisconnectedCallback,
+            <String, dynamic>{'handle': 0}),
+      );
+      expect(stubPlatform.countPreviousCalls(startConnectionCall), equals(2));
+    });
+
+    test(
+        're-connects when operation returns BillingResponse.clientDisconnected',
+        () async {
+      final Map<String, dynamic> okValue = buildBillingResultMap(
+          const BillingResultWrapper(responseCode: BillingResponse.ok));
+      stubPlatform.addResponse(
+        name: acknowledgePurchaseCall,
+        value: buildBillingResultMap(
+          const BillingResultWrapper(
+            responseCode: BillingResponse.serviceDisconnected,
+          ),
+        ),
+      );
+      stubPlatform.addResponse(
+        name: startConnectionCall,
+        value: okValue,
+        additionalStepBeforeReturn: (dynamic _) => stubPlatform.addResponse(
+            name: acknowledgePurchaseCall, value: okValue),
+      );
+      final PurchaseDetails purchase =
+          GooglePlayPurchaseDetails.fromPurchase(dummyUnacknowledgedPurchase);
+      final BillingResultWrapper result =
+          await iapAndroidPlatform.completePurchase(purchase);
+      expect(
+        stubPlatform.countPreviousCalls(acknowledgePurchaseCall),
+        equals(2),
+      );
+      expect(stubPlatform.countPreviousCalls(startConnectionCall), equals(2));
+      expect(result.responseCode, equals(BillingResponse.ok));
+    });
   });
 
   group('isAvailable', () {
@@ -314,7 +357,7 @@ void main() {
                 }
               ]
             });
-            iapAndroidPlatform.billingClient.callHandler(call);
+            iapAndroidPlatform.billingClientManager.client.callHandler(call);
           });
       final Completer<PurchaseDetails> completer = Completer<PurchaseDetails>();
       PurchaseDetails purchaseDetails;
@@ -358,7 +401,7 @@ void main() {
               'responseCode': const BillingResponseConverter().toJson(sentCode),
               'purchasesList': const <dynamic>[]
             });
-            iapAndroidPlatform.billingClient.callHandler(call);
+            iapAndroidPlatform.billingClientManager.client.callHandler(call);
           });
       final Completer<PurchaseDetails> completer = Completer<PurchaseDetails>();
       PurchaseDetails purchaseDetails;
@@ -416,7 +459,7 @@ void main() {
                 }
               ]
             });
-            iapAndroidPlatform.billingClient.callHandler(call);
+            iapAndroidPlatform.billingClientManager.client.callHandler(call);
           });
       final Completer<String> consumeCompleter = Completer<String>();
       // adding call back for consume purchase
@@ -531,7 +574,7 @@ void main() {
                 }
               ]
             });
-            iapAndroidPlatform.billingClient.callHandler(call);
+            iapAndroidPlatform.billingClientManager.client.callHandler(call);
           });
       final Completer<String> consumeCompleter = Completer<String>();
       // adding call back for consume purchase
@@ -609,7 +652,7 @@ void main() {
                 }
               ]
             });
-            iapAndroidPlatform.billingClient.callHandler(call);
+            iapAndroidPlatform.billingClientManager.client.callHandler(call);
           });
       final Completer<String?> consumeCompleter = Completer<String?>();
       // adding call back for consume purchase
@@ -675,7 +718,7 @@ void main() {
                 }
               ]
             });
-            iapAndroidPlatform.billingClient.callHandler(call);
+            iapAndroidPlatform.billingClientManager.client.callHandler(call);
           });
       final Completer<String> consumeCompleter = Completer<String>();
       // adding call back for consume purchase
@@ -733,7 +776,7 @@ void main() {
               'responseCode': const BillingResponseConverter().toJson(sentCode),
               'purchasesList': const <dynamic>[]
             });
-            iapAndroidPlatform.billingClient.callHandler(call);
+            iapAndroidPlatform.billingClientManager.client.callHandler(call);
           });
 
       final Completer<PurchaseDetails> completer = Completer<PurchaseDetails>();
diff --git a/packages/in_app_purchase/in_app_purchase_android/test/stub_in_app_purchase_platform.dart b/packages/in_app_purchase/in_app_purchase_android/test/stub_in_app_purchase_platform.dart
index 75972e644fa..35e2807bc3b 100644
--- a/packages/in_app_purchase/in_app_purchase_android/test/stub_in_app_purchase_platform.dart
+++ b/packages/in_app_purchase/in_app_purchase_android/test/stub_in_app_purchase_platform.dart
@@ -5,7 +5,9 @@
 import 'dart:async';
 import 'package:flutter/services.dart';
 
-typedef AdditionalSteps = void Function(dynamic args);
+// `FutureOr<dynamic>` instead of `FutureOr<void>` to avoid
+// "don't assign to void" warnings.
+typedef AdditionalSteps = FutureOr<dynamic> Function(dynamic args);
 
 class StubInAppPurchasePlatform {
   final Map<String, dynamic> _expectedCalls = <String, dynamic>{};
@@ -36,7 +38,7 @@ class StubInAppPurchasePlatform {
     _previousCalls.add(call);
     if (_expectedCalls.containsKey(call.method)) {
       if (_additionalSteps[call.method] != null) {
-        _additionalSteps[call.method]!(call.arguments);
+        await _additionalSteps[call.method]!(call.arguments);
       }
       return Future<dynamic>.sync(() => _expectedCalls[call.method]);
     } else {