diff --git a/tests/rptest/services/cloud_cluster_utils.py b/tests/rptest/services/cloud_cluster_utils.py
new file mode 100644
index 0000000000000..8b8b719c30359
--- /dev/null
+++ b/tests/rptest/services/cloud_cluster_utils.py
@@ -0,0 +1,59 @@
+from rptest.clients.rpk import RpkTool
+
+
+class FakePanda:
+    logger = None
+
+    def __init__(self, context, log):
+        self._context = context
+        self.logger = log
+
+
+class CloudClusterUtils:
+    def __init__(self, context, logger):
+        # Create fake redpanda class with logger only
+        self.fake_panda = FakePanda(context, logger)
+        # Create rpk to use several functions that is isolated
+        # from actual redpanda service
+        self.rpk = RpkTool(self.fake_panda)
+        self.logger = logger
+
+    def _parse_plugin_list(self, plist):
+        """
+        Parses 'rpk plugin list' output
+        """
+        # 'NAME  PATH                                  SHADOWS\nbyoc  /home/ubuntu/.local/bin/.rpk.ac-byoc  \n'
+        _lines = plist.split('\n')
+        _headers = []
+        _plugins = []
+        for _line in _lines:
+            # cleanup repeated spaces
+            _l = ' '.join(_line.split())
+            # Get nice list
+            _fields = _l.lower().split()
+            if not _fields:
+                continue
+            elif _fields[0] == 'name':
+                _headers = _l.lower().split()
+            elif not _headers:
+                self.logger.warning(f"Error parsing rpk plugin list: {plist}")
+            else:
+                # Create dict
+                _p = {}
+                for idx in range(len(_fields)):
+                    _p[_headers[idx]] = _fields[idx]
+                _plugins.append(_p)
+        return _plugins
+
+    def rpk_plugin_install(self, plugin_name, reinstall=False):
+        _plist = self.rpk.plugin_list()
+        # parse plugin list
+        _plist = self._parse_plugin_list(_plist)
+        _installed = [p['name'] == plugin_name for p in _plist]
+        if any(_installed) and reinstall:
+            # uninstall if plugin present. Use sudo just in case
+            self.rpk._execute(
+                [self.rpk._rpk_binary(), 'plugin', 'uninstall', plugin_name])
+        # Install latest plugin
+        self.rpk._execute(
+            [self.rpk._rpk_binary(), 'plugin', 'install', plugin_name])
diff --git a/tests/rptest/services/redpanda.py b/tests/rptest/services/redpanda.py
index 227b8f27e89a0..8e9cf5c098627 100644
--- a/tests/rptest/services/redpanda.py
+++ b/tests/rptest/services/redpanda.py
@@ -1471,6 +1471,7 @@ def __init__(self,
         #     self.GLOBAL_CLOUD_PEER_OWNER_ID, None)
 
         self._cloud_cluster = CloudCluster(
+            context,
             self.logger,
             self._cc_config,
             provider_config=self._provider_config)
diff --git a/tests/rptest/services/redpanda_cloud.py b/tests/rptest/services/redpanda_cloud.py
index 9fae8fb4d87ca..8a2ef0975f2c3 100644
--- a/tests/rptest/services/redpanda_cloud.py
+++ b/tests/rptest/services/redpanda_cloud.py
@@ -9,6 +9,7 @@
 from enum import Enum
 from typing import Optional
 from ducktape.utils.util import wait_until
+from rptest.services.cloud_cluster_utils import CloudClusterUtils
 from rptest.services.provider_clients import make_provider_client
 from rptest.services.provider_clients.ec2_client import RTBS_LABEL
 
@@ -266,7 +267,7 @@ class CloudClusterConfig:
     type: str = "FMC"
     network: str = "public"
     config_profile_name: str = "default"
-
+    use_same_cluster: bool = True
     install_pack_ver: str = "latest"
     install_pack_url_template: str = ""
     install_pack_auth_type: str = ""
@@ -318,6 +319,7 @@ class CloudCluster():
     CHECK_BACKOFF_SEC = 60.0
 
     def __init__(self,
+                 context,
                  logger,
                  cluster_config,
                  delete_namespace=False,
@@ -400,6 +402,11 @@ def __init__(self,
                 raise RuntimeError("Private networking is not implemented "
                                    f"for '{self.config.provider}'")
 
+        # prepare rpk plugin
+        self.utils = CloudClusterUtils(context, self._logger)
+        if self.config.type == CLOUD_TYPE_BYOC:
+            self.utils.rpk_plugin_install('byoc', reinstall=True)
+
     @property
     def cluster_id(self):
         """