diff --git a/e2e/keywords/common.resource b/e2e/keywords/common.resource
index 817ac1e9db..271447b499 100644
--- a/e2e/keywords/common.resource
+++ b/e2e/keywords/common.resource
@@ -48,6 +48,7 @@ Cleanup test resources
     cleanup_backups
     cleanup_disks
     cleanup_backing_images
+    cleanup_engine_images
     reset_backupstore
 
 Cleanup test resources include off nodes
diff --git a/e2e/keywords/engine_image.resource b/e2e/keywords/engine_image.resource
new file mode 100644
index 0000000000..8ed4a7f6a4
--- /dev/null
+++ b/e2e/keywords/engine_image.resource
@@ -0,0 +1,9 @@
+*** Settings ***
+Documentation       Longhorn engine image related keywords
+
+Library             ../libs/keywords/engine_image_keywords.py
+
+*** Keywords ***
+Create compatible engine image
+    ${compatible_engine_image_name} =    deploy_compatible_engine_image
+    Set Test Variable    ${compatible_engine_image_name}
diff --git a/e2e/keywords/volume.resource b/e2e/keywords/volume.resource
index 31fdd599d5..085188f074 100644
--- a/e2e/keywords/volume.resource
+++ b/e2e/keywords/volume.resource
@@ -255,3 +255,8 @@ Record volume ${volume_id} replica names
 Check volume ${volume_id} replica names are as recorded
     ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
     check_volume_replica_names_recorded    ${volume_name}
+
+Upgrade volume ${volume_id} engine to compatible engine image
+    ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
+    upgrade_engine_image    ${volume_name}    ${compatible_engine_image_name}
+    wait_for_engine_image_upgrade_completed    ${volume_name}    ${compatible_engine_image_name}
diff --git a/e2e/libs/engine_image/__init__.py b/e2e/libs/engine_image/__init__.py
new file mode 100644
index 0000000000..e723e4f715
--- /dev/null
+++ b/e2e/libs/engine_image/__init__.py
@@ -0,0 +1 @@
+from engine_image.engine_image import EngineImage
diff --git a/e2e/libs/engine_image/engine_image.py b/e2e/libs/engine_image/engine_image.py
new file mode 100644
index 0000000000..a6cb7f1776
--- /dev/null
+++ b/e2e/libs/engine_image/engine_image.py
@@ -0,0 +1,105 @@
+import time
+from utility.utility import logging
+from utility.utility import get_longhorn_client
+from utility.utility import get_retry_count_and_interval
+
+
+class EngineImage():
+
+    UPGRADE_TEST_IMAGE_PREFIX = "longhornio/longhorn-test:upgrade-test"
+
+    def __init__(self):
+        self.retry_count, self.retry_interval = get_retry_count_and_interval()
+
+    def get_default_engine_image(self):
+        images = get_longhorn_client().list_engine_image()
+        for img in images:
+            if img.default:
+                return img
+        assert False, f"Failed to get default engine image: {images}"
+
+    def get_default_engine_image_versions(self):
+        default_img = self.get_default_engine_image()
+        cli_v = default_img.cliAPIVersion
+        cli_minv = default_img.cliAPIMinVersion
+        ctl_v = default_img.controllerAPIVersion
+        ctl_minv = default_img.controllerAPIMinVersion
+        data_v = default_img.dataFormatVersion
+        data_minv = default_img.dataFormatMinVersion
+        return cli_v, cli_minv, ctl_v, ctl_minv, data_v, data_minv
+
+    def deploy_compatible_engine_image(self):
+        cli_v, cli_minv, ctl_v, ctl_minv, data_v, data_minv = self.get_default_engine_image_versions()
+        compatible_engine_image_name = \
+            f"{self.UPGRADE_TEST_IMAGE_PREFIX}.{cli_v}-{cli_minv}.{ctl_v}-{ctl_minv}.{data_v}-{data_minv}"
+        image = self.create_engine_image(compatible_engine_image_name)
+        return image.image
+
+    def create_engine_image(self, image_name):
+        image = get_longhorn_client().create_engine_image(image=image_name)
+        image = self.wait_for_engine_image_deployed(image.name)
+        assert image.refCount == 0, f"Expected new engine image {image_name} refCount == 0, but it's {image.refCount}"
+        assert image.noRefSince != "", f"Expected new engine image {image_name} noRefSince not empty, but it's {image.noRefSince}"
+        return image
+
+    def wait_for_engine_image_deployed(self, image_name):
+        self.wait_for_engine_image_created(image_name)
+        for i in range(self.retry_count):
+            image = get_longhorn_client().by_id_engine_image(image_name)
+            if image.state == "deployed":
+                break
+            time.sleep(self.retry_interval)
+        assert image.state == "deployed", f"Failed to deploy engine image {image_name}: {image}"
+        return image
+
+    def wait_for_engine_image_ref_count(self, image_name, count):
+        self.wait_for_engine_image_created(image_name)
+        for i in range(self.retry_count):
+            image = get_longhorn_client().by_id_engine_image(image_name)
+            if image.refCount == count:
+                break
+            time.sleep(self.retry_interval)
+        assert image.refCount == count, f"Failed to wait engine image {image_name} reference count {count}: {image}"
+        if count == 0:
+            assert image.noRefSince != "", f"Expected engine image {image_name} noRefSince non-empty: {image}"
+        return image
+
+    def wait_for_engine_image_created(self, image_name):
+        for i in range(self.retry_count):
+            images = get_longhorn_client().list_engine_image()
+            found = False
+            for img in images:
+                if img.name == image_name:
+                    found = True
+                    break
+            if found:
+                break
+            time.sleep(self.retry_interval)
+        assert found, f"Failed to create engine image {image_name}: {images}"
+
+    def cleanup_engine_images(self):
+        images = get_longhorn_client().list_engine_image().data
+        for image in images:
+            if not image.default:
+                self.wait_for_engine_image_ref_count(image.name, 0)
+                get_longhorn_client().delete(image)
+                self.wait_for_engine_image_deleted(image.name)
+
+    def wait_for_engine_image_deleted(self, image_name):
+
+        deleted = False
+
+        for i in range(self.retry_count):
+
+            time.sleep(self.retry_interval)
+            deleted = True
+
+            images = get_longhorn_client().list_engine_image().data
+            for image in images:
+                if image.name == image_name:
+                    deleted = False
+                    break
+            if deleted:
+                break
+
+        assert deleted, f"Failed to delete engine image {image_name}: {images}"
diff --git a/e2e/libs/keywords/engine_image_keywords.py b/e2e/libs/keywords/engine_image_keywords.py
new file mode 100644
index 0000000000..dd84f05b01
--- /dev/null
+++ b/e2e/libs/keywords/engine_image_keywords.py
@@ -0,0 +1,13 @@
+from engine_image import EngineImage
+from utility.utility import logging
+
+class engine_image_keywords:
+
+    def __init__(self):
+        self.engine_image = EngineImage()
+
+    def deploy_compatible_engine_image(self):
+        return self.engine_image.deploy_compatible_engine_image()
+
+    def cleanup_engine_images(self):
+        return self.engine_image.cleanup_engine_images()
diff --git a/e2e/libs/keywords/volume_keywords.py b/e2e/libs/keywords/volume_keywords.py
index 0be59ce1ef..726f693e4d 100644
--- a/e2e/libs/keywords/volume_keywords.py
+++ b/e2e/libs/keywords/volume_keywords.py
@@ -305,3 +305,9 @@ def check_volume_replica_names_recorded(self, volume_name):
             f"Volume {volume_name} replica names mismatched:\n" \
             f"Want: {expected_replica_names}\n" \
             f"Got: {actual_replica_names}"
+
+    def upgrade_engine_image(self, volume_name, engine_image_name):
+        self.volume.upgrade_engine_image(volume_name, engine_image_name)
+
+    def wait_for_engine_image_upgrade_completed(self, volume_name, engine_image_name):
+        self.volume.wait_for_engine_image_upgrade_completed(volume_name, engine_image_name)
diff --git a/e2e/libs/volume/crd.py b/e2e/libs/volume/crd.py
index 77e11df9bb..41fbb4fcf3 100644
--- a/e2e/libs/volume/crd.py
+++ b/e2e/libs/volume/crd.py
@@ -53,7 +53,9 @@ def create(self, volume_name, size, numberOfReplicas, frontend, migratable, acce
                 "dataEngine": dataEngine,
                 "backingImage": backingImage,
                 "Standby": Standby,
-                "fromBackup": fromBackup
+                "fromBackup": fromBackup,
+                # disable revision counter by default from v1.7.0
+                "revisionCounterDisabled": True
             }
         }
         try:
@@ -487,3 +489,9 @@ def create_persistentvolume(self, volume_name, retry):
 
     def create_persistentvolumeclaim(self, volume_name, retry):
         return Rest(self.node_exec).create_persistentvolumeclaim(volume_name, retry)
+
+    def upgrade_engine_image(self, volume_name, engine_image_name):
+        return Rest(self.node_exec).upgrade_engine_image(volume_name, engine_image_name)
+
+    def wait_for_engine_image_upgrade_completed(self, volume_name, engine_image_name):
+        return Rest(self.node_exec).wait_for_engine_image_upgrade_completed(volume_name, engine_image_name)
diff --git a/e2e/libs/volume/rest.py b/e2e/libs/volume/rest.py
index a628df2279..6f1a80b027 100644
--- a/e2e/libs/volume/rest.py
+++ b/e2e/libs/volume/rest.py
@@ -315,3 +315,36 @@ def create_persistentvolumeclaim(self, volume_name, retry):
                 break
             time.sleep(self.retry_interval)
         assert created
+
+    def upgrade_engine_image(self, volume_name, engine_image_name):
+        volume = self.get(volume_name)
+        volume.engineUpgrade(image=engine_image_name)
+
+    def wait_for_engine_image_upgrade_completed(self, volume_name, engine_image_name):
+        for i in range(self.retry_count):
+            volume = self.get(volume_name)
+            if volume.currentImage == engine_image_name:
+                break
+            time.sleep(self.retry_interval)
+        assert volume.currentImage == engine_image_name, f"Failed to upgrade engine image to {engine_image_name}: {volume}"
+        self.wait_for_replica_ready_to_rw(volume_name)
+
+    def wait_for_replica_ready_to_rw(self, volume_name):
+        for _ in range(self.retry_count):
+            ready = True
+            volume = self.get(volume_name)
+            replicas = volume.replicas
+            if len(replicas) != volume.numberOfReplicas:
+                logging(f"Waiting for volume {volume_name} replica count = {volume.numberOfReplicas}, current count = {len(replicas)}")
+                ready = False
+                time.sleep(self.retry_interval)
+                continue
+            else:
+                for replica in replicas:
+                    if replica.mode != "RW":
+                        ready = False
+                        break
+            if ready:
+                break
+            time.sleep(self.retry_interval)
+        assert ready, f"Failed to get volume {volume_name} replicas ready: {replicas}"
diff --git a/e2e/libs/volume/volume.py b/e2e/libs/volume/volume.py
index 88408f8aa6..7d8fcec34b 100644
--- a/e2e/libs/volume/volume.py
+++ b/e2e/libs/volume/volume.py
@@ -145,3 +145,9 @@ def create_persistentvolume(self, volume_name, retry):
 
     def create_persistentvolumeclaim(self, volume_name, retry):
         return self.volume.create_persistentvolumeclaim(volume_name, retry)
+
+    def upgrade_engine_image(self, volume_name, engine_image_name):
+        return self.volume.upgrade_engine_image(volume_name, engine_image_name)
+
+    def wait_for_engine_image_upgrade_completed(self, volume_name, engine_image_name):
+        return self.volume.wait_for_engine_image_upgrade_completed(volume_name, engine_image_name)
diff --git a/e2e/tests/regression/test_engine_image.robot b/e2e/tests/regression/test_engine_image.robot
new file mode 100644
index 0000000000..ed0e2dbdd7
--- /dev/null
+++ b/e2e/tests/regression/test_engine_image.robot
@@ -0,0 +1,32 @@
+*** Settings ***
+Documentation    Engine Image Test Cases
+
+Test Tags    regression    engine_image
+
+Resource    ../keywords/common.resource
+Resource    ../keywords/volume.resource
+Resource    ../keywords/engine_image.resource
+
+Test Setup    Set test environment
+Test Teardown    Cleanup test resources
+
+*** Variables ***
+${LOOP_COUNT}    1
+${RETRY_COUNT}    300
+${RETRY_INTERVAL}    1
+${DATA_ENGINE}    v1
+
+*** Test Cases ***
+Test Replica Rebuilding After Engine Upgrade
+    [Tags]    coretest
+    Given Create compatible engine image
+    And Create volume 0
+    And Attach volume 0
+    And Wait for volume 0 healthy
+    And Write data to volume 0
+    When Upgrade volume 0 engine to compatible engine image
+    Then Delete volume 0 replica on node 1
+    And Wait until volume 0 replica rebuilding started on node 1
+    And Wait until volume 0 replica rebuilding completed on node 1
+    And Wait for volume 0 healthy
+    And Check volume 0 data is intact