diff --git a/gpMgmt/bin/gppylib/commands/gp.py b/gpMgmt/bin/gppylib/commands/gp.py
index b980bc629c4..64aceee7e1e 100644
--- a/gpMgmt/bin/gppylib/commands/gp.py
+++ b/gpMgmt/bin/gppylib/commands/gp.py
@@ -858,7 +858,18 @@ class ModifyConfSetting(Command):
     def __init__(self, name, file, optName, optVal, optType='string', ctxt=LOCAL, remoteHost=None):
         cmdStr = None
         if optType == 'number':
-            cmdStr = "perl -p -i.bak -e 's/^%s[ ]*=[ ]*\\d+/%s=%d/' %s" % (optName, optName, optVal, file)
+            # Replace the whole value: valid integers can be signed, octal,
+            # hexadecimal, or quoted scientific notation. Matching only the
+            # initial digits can leave a suffix that corrupts the config.
+            # Keep trailing whitespace/comments (including CRLF line endings)
+            # and require a setting-name
+            # separator so similarly named parameters are not changed.
+            # \x27 is a quote inside the shell's single-quoted Perl program.
+            cmdStr = (
+                "perl -p -i.bak -e 's/^[ \\t]*%s(?:[ \\t]*=[ \\t]*|[ \\t]+)"
+                "(?:\\x27[^\\x27]*\\x27|[^\\s\\x27#]+)"
+                "(?=[ \\t]*(?:#|\\r?$))/%s=%d/i' %s"
+            ) % (optName, optName, optVal, file)
         elif optType == 'string':
             cmdStr = "perl -i -p -e \"s/^%s[ ]*=[ ]*'[^']*'/%s='%s'/\" %s" % (optName, optName, optVal, file)
         else:
@@ -1670,6 +1681,22 @@ def createTempDirectoryName(coordinatorDataDirectory, tempDirPrefix):
                                 datetime.datetime.now().strftime('%m%d%Y'),
                                 os.getpid())
 
+# Cherry-picked from Greenplum Database (greenplum-db/gpdb-archive @ 482967c1b4),
+# gpMgmt/bin/gppylib/commands/gp.py.
+def is_gprecoverseg_running():
+    gprecoverseg_pidfile = os.path.join(get_coordinatordatadir(), 'gprecoverseg.lock', 'PID')
+    try:
+        with open(gprecoverseg_pidfile) as pidfile:
+            # The file can still be empty while the lock owner writes its PID.
+            gprecoverseg_pid = int(pidfile.read())
+    except (OSError, ValueError):
+        return False
+
+    if gprecoverseg_pid <= 0:
+        return False
+
+    return check_pid(gprecoverseg_pid)
+
 #-------------------------------------------------------------------------
 class GpRecoverSeg(Command):
    """
diff --git a/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py b/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py
index 4ac75c69685..9ba728d1a8e 100644
--- a/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py
+++ b/gpMgmt/bin/gppylib/commands/test/unit/test_unit_gp.py
@@ -4,9 +4,9 @@
 #
 
 from gppylib.commands.base import CommandResult
-from mock import patch
+from mock import patch, mock_open
 
-from gppylib.commands.gp import is_pid_postmaster, get_postmaster_pid_locally
+from gppylib.commands.gp import is_pid_postmaster, get_postmaster_pid_locally, is_gprecoverseg_running
 from test.unit.gp_unittest import GpTestCase, run_tests
 
 
@@ -158,5 +158,38 @@ def test_get_postmaster_pid_locally_error(self, mock1):
     def test_get_postmaster_pid_locally_empty(self, mock1):
         self.assertEqual(get_postmaster_pid_locally('/tmp'), -1)
 
+    @patch('gppylib.commands.gp.check_pid', return_value=True)
+    @patch('gppylib.commands.gp.get_coordinatordatadir', return_value='/data/coordinator')
+    @patch("builtins.open", new_callable=mock_open, read_data="123")
+    def test_is_gprecoverseg_running_succeeds(self, mock_file, mock1, mock2):
+        result = is_gprecoverseg_running()
+        mock_file.assert_called_once_with('/data/coordinator/gprecoverseg.lock/PID')
+        mock2.assert_called_once_with(123)
+        self.assertTrue(result)
+
+    @patch('gppylib.commands.gp.check_pid')
+    @patch('gppylib.commands.gp.get_coordinatordatadir', return_value='/data/coordinator')
+    def test_is_gprecoverseg_running_with_invalid_pid(self, mock_datadir, mock_check_pid):
+        for contents in ['', ' \n', 'not-a-pid', '0', '-1']:
+            with self.subTest(contents=contents):
+                with patch('builtins.open', mock_open(read_data=contents)):
+                    self.assertFalse(is_gprecoverseg_running())
+        mock_check_pid.assert_not_called()
+
+    @patch('gppylib.commands.gp.check_pid', return_value=False)
+    @patch('gppylib.commands.gp.get_coordinatordatadir', return_value='/data/coordinator')
+    @patch('builtins.open', new_callable=mock_open, read_data='123')
+    def test_is_gprecoverseg_running_with_dead_pid(self, mock_file, mock_datadir, mock_check_pid):
+        self.assertFalse(is_gprecoverseg_running())
+        mock_check_pid.assert_called_once_with(123)
+
+    @patch('gppylib.commands.gp.check_pid')
+    @patch('gppylib.commands.gp.get_coordinatordatadir', return_value='/invalid/path/')
+    def test_is_gprecoverseg_running_when_pidfile_does_not_exists(self, mock1, mock2):
+        result = is_gprecoverseg_running()
+        self.assertFalse(result)
+        self.assertFalse(mock2.called)
+
+
 if __name__ == '__main__':
     run_tests()
diff --git a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py
index 869a37d312b..1c5997833ae 100644
--- a/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py
+++ b/gpMgmt/bin/gppylib/operations/buildMirrorSegments.py
@@ -240,6 +240,9 @@ def __build_mirrors(self, actionName, gpEnv, gpArray):
 
         recovery_info_by_host = recoveryinfo.build_recovery_info(self.__mirrorsToBuild)
 
+        # Remove any existing progress files for segments to be recovered
+        self.remove_existing_progress_files(recovery_info_by_host)
+
         self._run_setup_recovery(actionName, recovery_info_by_host)
 
         backout_map = self._update_config(recovery_info_by_host, gpArray)
@@ -293,13 +296,17 @@ def _update_config(self, recovery_info_by_host, gpArray):
 
         return backout_map
 
-    def _remove_progress_files(self, recovery_info_by_host, recovery_results):
+    # Remove any existing progress file of segments that will be recovered by
+    # current gprecoverseg execution. The files written by this run are left in
+    # place: gpstate reads them to report progress, and they are what an
+    # operator looks at after a recovery that went wrong.
+    def remove_existing_progress_files(self, recovery_info_by_host):
         remove_progress_file_cmds = []
         for hostName, recovery_info_list in recovery_info_by_host.items():
             for ri in recovery_info_list:
-                if recovery_results.was_bb_rewind_successful(ri.target_segment_dbid):
-                    remove_progress_file_cmds.append(self._get_remove_cmd(ri.progress_file, hostName))
-        self.__runWaitAndCheckWorkerPoolForErrorsAndClear(remove_progress_file_cmds, suppressErrorCheck=False)
+                remove_progress_file_cmds.append(self._get_remove_cmd("*dbid{}.out".format(ri.target_segment_dbid),
+                                                                      hostName))
+        self.__runWaitAndCheckWorkerPoolForErrorsAndClear(remove_progress_file_cmds, suppressErrorCheck=True)
 
     def _revert_config_update(self, recovery_results, backout_map):
         if len(backout_map) == 0:
@@ -447,8 +454,12 @@ def _get_progress_cmd(self, progressFile, targetSegmentDbId, targetHostname):
                                                        remoteHost=targetHostname)
         return None
 
+    # remove_file is a name pattern (e.g. "*dbid2.out"), matched in the log
+    # directory the progress files are written to.
     def _get_remove_cmd(self, remove_file, target_host):
-        return base.Command("remove file", "rm -f {}".format(shlex.quote(remove_file)), ctxt=base.REMOTE, remoteHost=target_host)
+        return base.Command("remove file", "find {} -name {} -delete".format(shlex.quote(gplog.get_logger_dir()),
+                                                                          shlex.quote(remove_file)),
+                            ctxt=base.REMOTE, remoteHost=target_host)
 
     def __runWaitAndCheckWorkerPoolForErrorsAndClear(self, cmds, suppressErrorCheck=False, progressCmds=[]):
         for cmd in cmds:
@@ -486,7 +497,6 @@ def _run_recovery(self, action_name, recovery_info_by_host, gpEnv):
         recovery_results = RecoveryResult(action_name, completed_recovery_results, self.__logger)
         recovery_results.print_bb_rewind_and_start_errors()
 
-        self._remove_progress_files(recovery_info_by_host, recovery_results)
         return recovery_results
 
     def _do_recovery(self, recovery_info_by_host, gpEnv):
diff --git a/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py
index 61a88c21539..0154b83d139 100644
--- a/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py
+++ b/gpMgmt/bin/gppylib/operations/test/unit/test_unit_buildMirrorSegments.py
@@ -522,6 +522,24 @@ def test_run_setup_recovery_empty_errors(self):
 
         self._assert_setup_recovery()
 
+    def test_remove_existing_progress_files(self):
+        self._setup_recovery_mocks()
+
+        recovery_info = OrderedDict()
+        recovery_info['host1'] = [self.recovery_info1]
+        recovery_info['host2'] = [self.recovery_info2, self.recovery_info3]
+
+        self.default_build_mirrors_obj.remove_existing_progress_files(recovery_info)
+
+        self.assertEqual([call(ANY, suppressErrorCheck=True)],
+                         self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list)
+        rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][0][0]
+        self.assertEqual(['host1', 'host2', 'host2'], [cmd.remoteHost for cmd in rm_progress_cmds])
+        self.assertEqual(["find /tmp/logdir -name '*dbid2.out' -delete",
+                          "find /tmp/logdir -name '*dbid3.out' -delete",
+                          "find /tmp/logdir -name '*dbid4.out' -delete"],
+                         [cmd.cmdStr for cmd in rm_progress_cmds])
+
     def test_run_recovery_no_errors(self):
         self._setup_recovery_mocks()
 
@@ -531,12 +549,11 @@ def test_run_recovery_no_errors(self):
 
         self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv)
 
-        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)],
+        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)],
                          self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list)
 
         seg_recovery_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][0][0]
         progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[0][1]['progressCmds']
-        rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0]
 
         #TODO fix formatting
         expected_recovery_cmd_strs = ["""$GPHOME/sbin/gpsegrecovery.py -c '[{"target_datadir": "/datadir2", "target_port": 7001, "target_segment_dbid": 2, "source_hostname": "source_host_for_dbid2", "source_port": 7002, "is_full_recovery": true, "progress_file": "/tmp/progress_file2"}]' -l /tmp/logdir -b 64 --era=dummy_era""", """$GPHOME/sbin/gpsegrecovery.py -c '[{"target_datadir": "/datadir3", "target_port": 7003, "target_segment_dbid": 3, "source_hostname": "source_host_for_dbid3", "source_port": 7004, "is_full_recovery": true, "progress_file": "/tmp/progress_file3"}, {"target_datadir": "/datadir4", "target_port": 7005, "target_segment_dbid": 4, "source_hostname": "source_host_for_dbid4", "source_port": 7006, "is_full_recovery": true, "progress_file": "/tmp/progress_file4"}]' -l /tmp/logdir -b 64 --era=dummy_era"""]
@@ -558,13 +575,6 @@ def test_run_recovery_no_errors(self):
                          progress_cmds[2].cmdStr)
         self.assertEqual(1, self.mock_gp_era.call_count)
 
-        self.assertEqual(3, len(rm_progress_cmds))
-        self.assertEqual("host1", rm_progress_cmds[0].remoteHost)
-        self.assertEqual("rm -f /tmp/progress_file2", rm_progress_cmds[0].cmdStr)
-        self.assertEqual("host2", rm_progress_cmds[1].remoteHost)
-        self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[1].cmdStr)
-        self.assertEqual("host2", rm_progress_cmds[2].remoteHost)
-        self.assertEqual("rm -f /tmp/progress_file4", rm_progress_cmds[2].cmdStr)
 
         self._assert_run_recovery()
 
@@ -584,11 +594,9 @@ def test_run_recovery_all_dbids_fail_only_bb_rewind_errors(self):
         self._setup_recovery_mocks([host1_error, host2_error])
 
         self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv)
-        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)],
+        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)],
                          self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list)
-        rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0]
 
-        self.assertEqual(0, len(rm_progress_cmds))
 
         self._assert_run_recovery()
 
@@ -599,13 +607,9 @@ def test_run_recovery_some_dbids_fail_all_bb_rewind_errors(self):
         self._setup_recovery_mocks([host1_error, host2_error])
 
         self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv)
-        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)],
+        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)],
                          self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list)
-        rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0]
 
-        self.assertEqual(1, len(rm_progress_cmds))
-        self.assertEqual("host2", rm_progress_cmds[0].remoteHost)
-        self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[0].cmdStr)
 
         self._assert_run_recovery()
 
@@ -616,11 +620,8 @@ def test_run_recovery_all_dbids_fail_all_start_errors(self):
         host2_error = '[{"error_type": "start", "error_msg":"some error for dbid 4", "dbid": 4, "datadir": "/datadir4", "port": 7005, "progress_file": "/tmp/progress4"}]'
         self._setup_recovery_mocks([host1_error, host2_error])
         self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv)
-        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)],
+        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)],
                          self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list)
-        rm_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0]
-
-        self.assertEqual(3, len(rm_cmds))
         self._assert_run_recovery()
 
     def test_run_recovery_some_dbids_fail_all_start_errors(self):
@@ -631,10 +632,8 @@ def test_run_recovery_some_dbids_fail_all_start_errors(self):
         self._setup_recovery_mocks([host1_error, host2_error])
 
         self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv)
-        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)],
+        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)],
                          self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list)
-        rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0]
-        self.assertEqual(3, len(rm_progress_cmds))
 
         self._assert_run_recovery()
 
@@ -647,14 +646,10 @@ def test_run_recovery_all_dbids_fail_bb_rewind_and_start_errors(self):
 
         self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv)
 
-        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)],
+        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)],
                          self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list)
-        rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0]
 
         # Since start passed for dbid:3, we should have deleted it's progress file
-        self.assertEqual(1, len(rm_progress_cmds))
-        self.assertEqual("host1", rm_progress_cmds[0].remoteHost)
-        self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[0].cmdStr)
 
         self._assert_run_recovery()
 
@@ -671,15 +666,9 @@ def test_run_recovery_all_dbids_fail_bb_rewind_and_default_errors(self):
 
         self.default_build_mirrors_obj._run_recovery(self.default_action_name, recovery_info, self.gpEnv)
 
-        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True), call(ANY, suppressErrorCheck=False)],
+        self.assertEqual([call(ANY, progressCmds=ANY, suppressErrorCheck=True)],
                          self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list)
-        rm_progress_cmds = self.default_build_mirrors_obj._GpMirrorListToBuild__runWaitAndCheckWorkerPoolForErrorsAndClear.call_args_list[1][0][0]
 
-        self.assertEqual(2, len(rm_progress_cmds))
-        self.assertEqual("host1", rm_progress_cmds[0].remoteHost)
-        self.assertEqual("host2", rm_progress_cmds[1].remoteHost)
-        self.assertEqual("rm -f /tmp/progress_file2", rm_progress_cmds[0].cmdStr)
-        self.assertEqual("rm -f /tmp/progress_file3", rm_progress_cmds[1].cmdStr)
 
         self._assert_run_recovery()
 
diff --git a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py
index 81377136f83..503360368a4 100644
--- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py
+++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment.py
@@ -262,6 +262,25 @@ def run(self):
         if optionCnt > 1:
             raise ProgramArgumentValidationException("Only one of -i, -p, and -r may be specified")
 
+        # -o only writes out a sample configuration file describing what would
+        # be recovered; it does not recover anything, so combining it with an
+        # option that says where to recover to is a contradiction rather than a
+        # refinement.
+        if self.__options.outputSampleConfigFile:
+            if self.__options.recoveryConfigFile is not None:
+                raise ProgramArgumentValidationException("Invalid -i provided with -o argument")
+            if self.__options.rebalanceSegments:
+                raise ProgramArgumentValidationException("Invalid -r provided with -o argument")
+
+        # A full resynchronisation rebuilds the mirror from its primary, which
+        # is neither what -r (rebalance back to preferred roles) nor -p
+        # (recover onto a different host) asks for.
+        if self.__options.forceFullResynchronization:
+            if self.__options.rebalanceSegments:
+                raise ProgramArgumentValidationException("-F option is not supported with -r option")
+            if self.__options.newRecoverHosts is not None:
+                raise ProgramArgumentValidationException("-F option is not supported with -p option")
+
         faultProberInterface.getFaultProber().initializeProber(gpEnv.getCoordinatorPort())
 
         confProvider = configInterface.getConfigurationProvider().initializeProvider(gpEnv.getCoordinatorPort())
@@ -378,12 +397,16 @@ def signal_handler(sig, frame):
             if not mirrorBuilder.recover_mirrors(gpEnv, gpArray):
                 if self.termination_requested:
                     self.logger.error("gprecoverseg process was interrupted by the user.")
+                self.logger.error("gprecoverseg failed. Please check the output for more details.")
                 sys.exit(1)
 
             if self.termination_requested:
                 self.logger.info("Not able to terminate the recovery process since it has been completed successfully.")
 
             self.logger.info("********************************")
+            self.logger.info("Future gprecoverseg executions might remove the currently created pg_basebackup/pg_rewind/rsync "
+                             "progress files, please save these files if needed.")
+            self.logger.info("********************************")
             self.logger.info("Segments successfully recovered.")
             self.logger.info("********************************")
 
diff --git a/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py b/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py
index 34b6a12ff2e..823a3dbc8e1 100644
--- a/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py
+++ b/gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py
@@ -1,11 +1,102 @@
 import abc
 from typing import List
 
+from contextlib import closing
+from gppylib.db import dbconn
+from gppylib import gplog
 from gppylib.mainUtils import ExceptionNoStackTraceNeeded
 from gppylib.operations.detect_unreachable_hosts import get_unreachable_segment_hosts
 from gppylib.parseutils import line_reader, check_values, canonicalize_address
-from gppylib.utils import checkNotNone, normalizeAndValidateInputPath
+from gppylib.utils import checkNotNone, normalizeAndValidateInputPath, validateHostnameAddress
 from gppylib.gparray import GpArray, Segment
+from gppylib.commands.gp import RECOVERY_REWIND_APPNAME
+
+logger = gplog.get_default_logger()
+
+
+# cherry-pick from Greenplum Database
+# greenplum-db/gpdb-archive @ 482967c1b4,
+# gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py
+def get_segments_with_running_basebackup(coordinator_port):
+    """
+    Returns a list of contentIds of source segments of running pg_basebackup processes
+    At present gp_stat_replication table does not contain any info about datadir and dbid of the target of running pg_basebackup
+
+    coordinator_port is passed in because DbURL() on its own uses PGPORT, or
+    5432 when that is unset, and neither need be the coordinator's port.
+    """
+
+    sql = "select gp_segment_id from gp_stat_replication where application_name = 'pg_basebackup'"
+
+    try:
+        with closing(dbconn.connect(dbconn.DbURL(port=coordinator_port))) as conn:
+            res = dbconn.query(conn, sql)
+            rows = res.fetchall()
+    except Exception as e:
+        raise Exception("Failed to query gp_stat_replication: %s" %str(e))
+
+    segments_with_running_basebackup = {row[0] for row in rows}
+
+    if len(segments_with_running_basebackup) == 0:
+        logger.debug("No basebackup running")
+    return segments_with_running_basebackup
+
+
+# cherry-pick from Greenplum Database
+# greenplum-db/gpdb-archive @ 482967c1b4,
+# gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py
+def is_pg_rewind_running(hostname, port):
+    """
+        Returns true if a pg_rewind process is running for the given segment
+    """
+    logger.debug(
+        "Checking for running instances of pg_rewind with host {} and port {} as source server".format(hostname, port))
+
+    """
+        Reasons to depend on pg_stat_activity table:
+            * pg_rewind is invoked using --source-server connection string as it needs libpq connection
+              with source server, which will be remote to the target server and --source-pgdata can not
+              be used in that case. Thus pg_stat_activity will always contain entry for active pg_rewind.
+            * pg_rewind keeps a connection open throughout it's lifecycle, so pg_stat_activity will contain
+              entries for active pg_rewind process till the end of execution.
+            * gpstate uses the above mentioned approach (pg_stat_activity entry check) to check for
+              incremental recoveries in progress.Thus, using the same approach will make it consistent
+              everywhere.
+    """
+
+    sql = "SELECT count(*) FROM pg_stat_activity WHERE application_name = '{}'".format(RECOVERY_REWIND_APPNAME)
+    try:
+        url = dbconn.DbURL(hostname=hostname, port=port, dbname='template1')
+        with closing(dbconn.connect(url, utility=True)) as conn:
+            res = dbconn.querySingleton(conn, sql)
+            return res > 0
+
+    except Exception as e:
+        raise Exception("Failed to query pg_stat_activity for segment hostname: {}, port: {}, error: {}".format(
+            hostname, str(port), str(e)))
+
+
+# cherry-pick from Greenplum Database
+# greenplum-db/gpdb-archive @ 482967c1b4,
+# gpMgmt/bin/gppylib/programs/clsRecoverSegment_triples.py
+def extract_recovery_config_info(parts):
+    """
+    Extracts relevant recovery configuration information from a list of parts.
+
+    Three parts is the historical form, "<address>|<port>|<datadir>", and the
+    hostname is taken to be the address. Four parts puts an explicit hostname
+    first; it is then cross-checked against the address, and against the
+    hostname the segment is registered under.
+    """
+    address, port, datadir = parts[-3:]
+    hostname_check_required = False
+    hostname = address
+
+    if len(parts) == 4:
+        hostname = parts[0]
+        hostname_check_required = True
+
+    return hostname, address, port, datadir, hostname_check_required
 
 
 class RecoveryTriplet:
@@ -84,10 +175,12 @@ def validate(failed, live, failover):
 
 
 class RecoveryTripletRequest:
-    def __init__(self, failed, failover_host=None, failover_port=None, failover_datadir=None, is_new_host=False):
+    def __init__(self, failed, failover_host_name=None, failover_host_address=None, failover_port=None,
+                 failover_datadir=None, is_new_host=False):
         self.failed = failed
 
-        self.failover_host = failover_host
+        self.failover_host_name = failover_host_name
+        self.failover_host_address = failover_host_address
         self.failover_port = failover_port
         self.failover_datadir = failover_datadir
         self.failover_to_new_host = is_new_host
@@ -142,12 +235,56 @@ def _convert_requests_to_triplets(self, requests: List[RecoveryTripletRequest])
         triplets = []
 
         dbIdToPeerMap = self.gpArray.getDbIdToPeerMap()
+
+        failed_segments_with_running_basebackup = []
+        failed_segments_with_running_pgrewind = []
+        segments_with_running_basebackup = get_segments_with_running_basebackup(
+            self.gpArray.coordinator.getSegmentPort())
+
         for req in requests:
+            """
+                When running gprecoverseg (any sort of recovery full/incremental), if the pg_rewind, pg_basebackup is
+                already running for a segment, that segment should be skipped from the recovery. The reason being that
+                there should be only one writer per target segment at a time. Having several writers to a target will
+                eventually make the segment inconsistent and in a weird state.
+
+                Although technically we could allow user to run a full recovery to a new host even if there is a
+                pg_rewind/pg_basebackup running for that segment. This is a pretty rare scenario and we have decided not
+                to over complicates the code just to support this scenario.
+            """
+            failed_segment_dbid = req.failed.getSegmentDbId()
+            peer = dbIdToPeerMap.get(failed_segment_dbid)
+            if peer is None:
+                raise Exception("No peer found for dbid {}. liveSegment is None".format(failed_segment_dbid))
+            peer_contentid = peer.getSegmentContentId()
+
+            # Every reason to skip this segment is settled before the failover
+            # block below, which rewrites the segment registered in gpArray.
+            # updateSystemConfig() writes every segment changed there to the
+            # catalog, so a skipped segment would be recorded at a location
+            # nothing was recovered to.
+            if req.failed.unreachable and not req.failover_to_new_host:
+                continue
+
+            # Asked only after the skip above. is_pg_rewind_running() connects
+            # to the peer, and on a double fault -- failed segment and peer
+            # both down -- that connection raises. Asked first, it would abort
+            # the whole gprecoverseg run rather than skip the one segment,
+            # which is the opposite of what is wanted when a cluster is already
+            # degraded.
+            if peer_contentid in segments_with_running_basebackup:
+                failed_segments_with_running_basebackup.append(peer_contentid)
+                continue
+
+            if is_pg_rewind_running(peer.getSegmentHostName(), peer.getSegmentPort()):
+                failed_segments_with_running_pgrewind.append(peer_contentid)
+                continue
+
             # TODO: These 2 cases have different behavior which might be confusing to the user.
             # "<failed_address>|<failed_port>|<failed_data_dir> <failed_address>|<failed_port>|<failed_data_dir>" does full recovery
             # "<failed_address>|<failed_port>|<failed_data_dir>" does incremental recovery
             failover = None
-            if req.failover_host:
+            if req.failover_host_address:
 
                 # these two lines make it so that failover points to the object that is registered in gparray
                 #   as the failed segment(!).
@@ -155,20 +292,32 @@ def _convert_requests_to_triplets(self, requests: List[RecoveryTripletRequest])
                 req.failed = failover.copy()
 
                 # now update values in failover segment
-                failover.setSegmentAddress(req.failover_host)
-                failover.setSegmentHostName(req.failover_host)
+                if req.failover_host_name != req.failover_host_address:
+                    # Validate if the hostname and address are of the same host
+                    if not validateHostnameAddress(req.failover_host_name, req.failover_host_address):
+                        logger.warning(
+                            "Not able to co-relate hostname:{0} with address:{1}. "
+                            "Skipping recovery for segments with contentId {2}"
+                            .format(req.failover_host_name, req.failover_host_address, peer_contentid))
+                        continue
+
+                failover.setSegmentHostName(req.failover_host_name)
+                failover.setSegmentAddress(req.failover_host_address)
                 failover.setSegmentPort(int(req.failover_port))
                 failover.setSegmentDataDirectory(req.failover_datadir)
                 failover.unreachable = False if req.failover_to_new_host else failover.unreachable
 
-            # this must come AFTER the if check above because failedSegment can be adjusted to
-            #   point to a different object
-            if req.failed.unreachable and not req.failover_to_new_host:
-                continue
+            triplets.append(RecoveryTriplet(req.failed, peer, failover))
 
-            peer = dbIdToPeerMap.get(req.failed.getSegmentDbId())
+        if len(failed_segments_with_running_basebackup) > 0:
+            logger.warning(
+                "Found pg_basebackup running for segments with contentIds %s, skipping recovery of these segments" % (
+                    failed_segments_with_running_basebackup))
 
-            triplets.append(RecoveryTriplet(req.failed, peer, failover))
+        if len(failed_segments_with_running_pgrewind) > 0:
+            logger.warning(
+                "Found pg_rewind running for segments with contentIds %s, skipping recovery of these segments" % (
+                    failed_segments_with_running_pgrewind))
 
         return triplets
 
@@ -222,7 +371,11 @@ def _check_new_hosts():
         for failedHost, failoverHost in zip(sorted(failedSegments.keys()), self.newHosts):
             for failed in failedSegments[failedHost]:
                 failoverPort = self.portAssigner.findAndReservePort(failoverHost, failoverHost)
-                req = RecoveryTripletRequest(failed, failoverHost, failoverPort, failed.getSegmentDataDirectory(), True)
+                req = RecoveryTripletRequest(failed, failover_host_name=failoverHost,
+                                             failover_host_address=failoverHost,
+                                             failover_port=failoverPort,
+                                             failover_datadir=failed.getSegmentDataDirectory(),
+                                             is_new_host=True)
                 requests.append(req)
 
         return self._convert_requests_to_triplets(requests)
@@ -285,6 +438,11 @@ def getTriplets(self):
         def _find_failed_from_row():
             failed = None
             for segment in self.gpArray.getDbList():
+                # When the input configuration file carries an explicit hostname
+                # it has to match the one the segment is registered under, or
+                # this is not the segment the operator meant.
+                if row['hostname_check_required'] and segment.getSegmentHostName() != row['failedHostname']:
+                    continue
                 if (segment.getSegmentAddress() == row['failedAddress']
                         and str(segment.getSegmentPort()) == row['failedPort']
                         and segment.getSegmentDataDirectory() == row['failedDataDirectory']):
@@ -292,15 +450,32 @@ def _find_failed_from_row():
                     break
 
             if failed is None:
-                raise Exception("A segment to recover was not found in configuration.  " \
-                                "This segment is described by address|port|directory '%s|%s|%s'" %
-                                (row['failedAddress'], row['failedPort'], row['failedDataDirectory']))
+                # Keep Cloudberry's wording -- the two spaces and the names of
+                # the fields -- rather than upstream's terser form. It tells
+                # the operator which field is which, and the scenario that
+                # covers this asserts on it literally.
+                msg = "A segment to recover was not found in configuration.  " \
+                      "This segment is described by "
+                if row['hostname_check_required']:
+                    msg += "hostname|address|port|directory '{}|{}|{}|{}'".format(
+                        row['failedHostname'], row['failedAddress'],
+                        row['failedPort'], row['failedDataDirectory'])
+                else:
+                    msg += "address|port|directory '{}|{}|{}'".format(
+                        row['failedAddress'], row['failedPort'],
+                        row['failedDataDirectory'])
+
+                raise Exception(msg)
 
             return failed
 
         requests = []
         for row in self.rows:
-            req = RecoveryTripletRequest(_find_failed_from_row(), row.get('newAddress'), row.get('newPort'), row.get('newDataDirectory'))
+            req = RecoveryTripletRequest(_find_failed_from_row(),
+                                         failover_host_name=row.get('newHostname'),
+                                         failover_host_address=row.get('newAddress'),
+                                         failover_port=row.get('newPort'),
+                                         failover_datadir=row.get('newDataDirectory'))
             requests.append(req)
 
         return self._convert_requests_to_triplets(requests)
@@ -322,31 +497,43 @@ def _parseConfigFile(config_file):
                     msg = "line %d of file %s: expected 1 or 2 groups but found %d" % (lineno, config_file, len(groups))
                     raise ExceptionNoStackTraceNeeded(msg)
                 parts = groups[0].split('|')
-                if len(parts) != 3:
-                    msg = "line %d of file %s: expected 3 parts on failed segment group, obtained %d" % (
+
+                if len(parts) not in {3, 4}:
+                    msg = "line {0} of file {1}: expected 3 or 4 parts on failed segment group, obtained {2}" .format(
                         lineno, config_file, len(parts))
                     raise ExceptionNoStackTraceNeeded(msg)
-                address, port, datadir = parts
-                check_values(lineno, address=address, port=port, datadir=datadir)
+
+                hostname, address, port, datadir, hostname_check_required = extract_recovery_config_info(parts)
+
+                check_values(lineno, hostname=hostname, address=address, port=port, datadir=datadir)
                 datadir = normalizeAndValidateInputPath(datadir, f.name, lineno)
 
                 row = {
+                    'failedHostname': hostname,
                     'failedAddress': address,
                     'failedPort': port,
                     'failedDataDirectory': datadir,
-                    'lineno': lineno
+                    'lineno': lineno,
+                    'hostname_check_required': hostname_check_required
                 }
                 if len(groups) == 2:
                     parts2 = groups[1].split('|')
-                    if len(parts2) != 3:
-                        msg = "line %d of file %s: expected 3 parts on new segment group, obtained %d" % (
-                            lineno, config_file, len(parts2))
+                    if len(parts2) not in [3, 4] or len(parts) != len(parts2):
+                        msg = "line {0} of file {1}: expected equal parts, either 3 or 4 on both segment group, obtained {2} on " \
+                              "group1 and {3} on group2" .format(
+                            lineno, config_file, len(parts), len(parts2))
                         raise ExceptionNoStackTraceNeeded(msg)
-                    address2, port2, datadir2 = parts2
-                    check_values(lineno, address=address2, port=port2, datadir=datadir2)
+                    if len(parts2) == 4:
+                        hostname2, address2, port2, datadir2 = parts2
+                    else:
+                        address2, port2, datadir2 = parts2
+                        hostname2 = address2
+
+                    check_values(lineno, hostname=hostname2, address=address2, port=port2, datadir=datadir2)
                     datadir2 = normalizeAndValidateInputPath(datadir2, f.name, lineno)
 
                     row.update({
+                        'newHostname': hostname2,
                         'newAddress': address2,
                         'newPort': port2,
                         'newDataDirectory': datadir2
diff --git a/gpMgmt/bin/gppylib/programs/clsSystemState.py b/gpMgmt/bin/gppylib/programs/clsSystemState.py
index effcaddd440..e21bc495378 100644
--- a/gpMgmt/bin/gppylib/programs/clsSystemState.py
+++ b/gpMgmt/bin/gppylib/programs/clsSystemState.py
@@ -22,7 +22,6 @@
 from gppylib.system import configurationInterface as configInterface
 from gppylib.system.environment import GpCoordinatorEnvironment
 from gppylib.utils import TableLogger
-from gppylib.commands.gp import get_coordinatordatadir
 
 logger = gplog.get_default_logger()
 
@@ -682,15 +681,18 @@ def logSegments(segments, logAsPairs, additionalFieldsToLog=[]):
 
         self.__addClusterDownWarning(gpArray, data)
 
-        recovery_progress_file = get_recovery_progress_file(gplog)
-        segments_under_recovery = self._parse_recovery_progress_data(data, recovery_progress_file, gpArray)
-        gprecoverseg_lock_dir = os.path.join(get_coordinatordatadir() + '/gprecoverseg.lock')
-        if segments_under_recovery and os.path.exists(gprecoverseg_lock_dir):
-            logger.info("----------------------------------------------------")
-            logger.info("Segments in recovery")
-            logSegments(segments_under_recovery, False, [VALUE_RECOVERY_TYPE, VALUE_RECOVERY_COMPLETED_BYTES, VALUE_RECOVERY_TOTAL_BYTES,
-                                                          VALUE_RECOVERY_PERCENTAGE])
-            exitCode = 1
+        # Check if gprecoverseg process is running. We do not want to
+        # show the progress if the gprecoverseg process is not running.
+        if gp.is_gprecoverseg_running():
+            recovery_progress_file = get_recovery_progress_file(gplog)
+            segments_under_recovery = self._parse_recovery_progress_data(data, recovery_progress_file, gpArray)
+
+            if segments_under_recovery:
+                logger.info("----------------------------------------------------")
+                logger.info("Segments in recovery")
+                logSegments(segments_under_recovery, False, [VALUE_RECOVERY_TYPE, VALUE_RECOVERY_COMPLETED_BYTES,
+                                                             VALUE_RECOVERY_TOTAL_BYTES, VALUE_RECOVERY_PERCENTAGE])
+                exitCode = 1
 
         # final output -- no errors, then log this message
         if exitCode == 0:
diff --git a/gpMgmt/bin/gppylib/programs/test/unit/test_cluster_clsrecoversegment_triples.py b/gpMgmt/bin/gppylib/programs/test/unit/test_cluster_clsrecoversegment_triples.py
index e2bf74bdf67..719fbd912ce 100644
--- a/gpMgmt/bin/gppylib/programs/test/unit/test_cluster_clsrecoversegment_triples.py
+++ b/gpMgmt/bin/gppylib/programs/test/unit/test_cluster_clsrecoversegment_triples.py
@@ -2,29 +2,44 @@
 
 import copy
 import io
-from mock import Mock, patch, MagicMock
+import itertools
+import os
+from mock import Mock, patch, MagicMock, call
 import tempfile
 
 import gppylib
 from gparray import Segment, GpArray
-from gppylib.programs.clsRecoverSegment_triples import RecoveryTripletsUserConfigFile, RecoveryTripletsFactory, RecoveryTriplet
-from test.unit.gp_unittest import GpTestCase
+from gppylib.programs.clsRecoverSegment_triples import RecoveryTripletsUserConfigFile, RecoveryTripletsFactory, \
+    RecoveryTriplet, get_segments_with_running_basebackup, is_pg_rewind_running, extract_recovery_config_info
+from test.unit.gp_unittest import GpTestCase, FakeCursor
 
 
 class RecoveryTripletsFactoryTestCase(GpTestCase):
     def setUp(self):
         # Set maxDiff to None to see the entire diff on the console in case of failure
         self.maxDiff = None
+        mock_logger = Mock(spec=['log', 'warn', 'info', 'debug', 'error', 'warning', 'fatal'])
+
+        self.apply_patches([
+            patch('gppylib.programs.clsRecoverSegment_triples.logger', return_value=mock_logger),
+        ])
+
+        self.mock_logger = self.get_mock_from_apply_patch('logger')
 
     def run_single_ConfigFile_test(self, test):
         with tempfile.NamedTemporaryFile() as f:
             f.write(test["config"].encode("utf-8"))
             f.flush()
-            return self._run_single_FromGpArray_test(test["gparray"], f.name, None, None, test.get("unreachable_existing_hosts"))
+            return self._run_single_FromGpArray_test(test["gparray"], f.name, None, None,
+                                                     test.get("is_pgrewind_running", itertools.repeat(False)),
+                                                     test.get("segments_with_running_basebackup", set()),
+                                                     test.get("unreachable_existing_hosts"))
 
     def run_single_GpArray_test(self, test):
         return self._run_single_FromGpArray_test(test["gparray"], None, test["new_hosts"], test.get("unreachable_hosts"),
-                                     test.get("unreachable_existing_hosts"))
+                                                 test.get("is_pgrewind_running", itertools.repeat(False)),
+                                                 test.get("segments_with_running_basebackup", set()),
+                                                 test.get("unreachable_existing_hosts"))
 
     #TODO: do we want new hosts here?  We do not officially support new hosts with "-i"
     def test_RecoveryTripletsUserConfigFile_getMirrorTriples_should_pass(self):
@@ -100,7 +115,91 @@ def test_RecoveryTripletsUserConfigFile_getMirrorTriples_should_pass(self):
                 "config": "sdw2|21000|/mirror/gpseg0",
                 "unreachable_existing_hosts": ['sdw2'],
                 "expected": []
-            }
+            },
+            # A skipped segment that the config file moves must keep its
+            # location in gpArray; run_pass_tests() checks that only the
+            # expected triplets changed it.
+            {
+                "name": "one_mirror_inconfig_has_running_basebackup",
+                "gparray": self.all_up_gparray_str,
+                "config": """sdw2|21000|/mirror/gpseg0 sdw2|21000|/mirror/gpseg0_new
+                             sdw1|21000|/mirror/gpseg6 sdw1|21000|/mirror/gpseg6_new""",
+                "segments_with_running_basebackup": {0},
+                "expected": [self._triplet('16|6|m|m|s|u|sdw1|sdw1|21000|/mirror/gpseg6',
+                                           '8|6|p|p|s|u|sdw4|sdw4|20000|/primary/gpseg6',
+                                           '16|6|m|m|s|u|sdw1|sdw1|21000|/mirror/gpseg6_new')]
+            },
+            {
+                "name": "some_mirrors_inconfig_have_running_basebackup",
+                "gparray": self.all_up_gparray_str,
+                "config": """sdw2|21000|/mirror/gpseg0 sdw2|21000|/mirror/gpseg0_new
+                             sdw2|21001|/mirror/gpseg1 sdw2|21000|/mirror/gpseg1_new
+                             sdw1|21000|/mirror/gpseg6 sdw1|21000|/mirror/gpseg6_new""",
+                "segments_with_running_basebackup": {0, 1},
+                "expected": [self._triplet('16|6|m|m|s|u|sdw1|sdw1|21000|/mirror/gpseg6',
+                                           '8|6|p|p|s|u|sdw4|sdw4|20000|/primary/gpseg6',
+                                           '16|6|m|m|s|u|sdw1|sdw1|21000|/mirror/gpseg6_new')]
+            },
+            {
+                "name": "all_mirrors_inconfig_have_running_basebackup",
+                "gparray": self.all_up_gparray_str,
+                "config": """sdw2|21000|/mirror/gpseg0 sdw2|21000|/mirror/gpseg0_new
+                             sdw2|21001|/mirror/gpseg1 sdw2|21000|/mirror/gpseg1_new
+                             sdw1|21000|/mirror/gpseg6 sdw1|21000|/mirror/gpseg6_new""",
+                "segments_with_running_basebackup": {0, 1, 6},
+                "expected": []
+            },
+            {
+                "name": "one_moved_mirror_inconfig_has_running_pgrewind",
+                "gparray": self.all_up_gparray_str,
+                "config": """sdw2|21000|/mirror/gpseg0 sdw2|21000|/mirror/gpseg0_new
+                             sdw1|21000|/mirror/gpseg6 sdw1|21000|/mirror/gpseg6_new""",
+                "is_pgrewind_running": [True, False],
+                "expected": [self._triplet('16|6|m|m|s|u|sdw1|sdw1|21000|/mirror/gpseg6',
+                                           '8|6|p|p|s|u|sdw4|sdw4|20000|/primary/gpseg6',
+                                           '16|6|m|m|s|u|sdw1|sdw1|21000|/mirror/gpseg6_new')]
+            },
+            {
+                "name": "one_mirror_inconfig_has_running_pgrewind",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "config": """sdw1|20000|/primary/gpseg0
+                             sdw1|20001|/primary/gpseg1
+                             sdw2|20000|/primary/gpseg2""",
+                "is_pgrewind_running": [True, False, False],
+                "expected": [self._triplet('3|1|m|p|s|d|sdw1|sdw1|20001|/primary/gpseg1',
+                                           '7|1|p|m|s|u|sdw2|sdw2|21001|/mirror/gpseg1',
+                                           None),
+                             self._triplet('4|2|m|p|s|d|sdw2|sdw2|20000|/primary/gpseg2',
+                                           '8|2|p|m|s|u|sdw3|sdw3|21000|/mirror/gpseg2',
+                                           None)]
+            },
+            {
+                "name": "mirrors_inconfig_have_running_pgrewind_and_pgbasebackup",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "config": """sdw1|20000|/primary/gpseg0
+                             sdw1|20001|/primary/gpseg1
+                             sdw2|20000|/primary/gpseg2""",
+                "segments_with_running_basebackup": {1, 2},
+                "is_pgrewind_running": [True, False, False],
+                "expected": []
+            },
+            {
+                "name": "all_mirrors_inconfig_have_running_pgrewind",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "config": """sdw1|20000|/primary/gpseg0
+                             sdw1|20001|/primary/gpseg1
+                             sdw2|20000|/primary/gpseg2""",
+                "is_pgrewind_running": [True, True, True],
+                "expected": []
+            },
+            {
+                "name": "in_place_1_part_with_4_parameter",
+                "gparray": self.all_up_gparray_str,
+                "config": "sdw2|sdw2|21000|/mirror/gpseg0",
+                "expected": [self._triplet('10|0|m|m|s|u|sdw2|sdw2|21000|/mirror/gpseg0',
+                                           '2|0|p|p|s|u|sdw1|sdw1|20000|/primary/gpseg0',
+                                           None)]
+            },
         ]
         self.run_pass_tests(tests, self.run_single_ConfigFile_test)
 
@@ -269,6 +368,70 @@ def test_RecoveryTripletsInPlaceAndNewHosts_getMirrorTriples_should_pass(self):
                                            '6|0|p|m|s|u|sdw2|sdw2|21000|/mirror/gpseg0',
                                            '2|0|m|p|s|d|new_1|new_1|20000|/primary/gpseg0', failed_unreachable=True)]
             },
+            {
+                "name": "one_failed_segments_has_running_basebackup",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "new_hosts": [],
+                "segments_with_running_basebackup": {0},
+                "expected": [self._triplet('3|1|m|p|s|d|sdw1|sdw1|20001|/primary/gpseg1',
+                                           '7|1|p|m|s|u|sdw2|sdw2|21001|/mirror/gpseg1',
+                                           None),
+                             self._triplet('4|2|m|p|s|d|sdw2|sdw2|20000|/primary/gpseg2',
+                                           '8|2|p|m|s|u|sdw3|sdw3|21000|/mirror/gpseg2',
+                                           None)]
+            },
+            {
+                "name": "all_failed_segments_have_running_basebackup",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "new_hosts": [],
+                "segments_with_running_basebackup": {0, 1, 2},
+                "expected": []
+            },
+            {
+                "name": "failed_segments_have_running_pgrewind_or_pgbasebackup",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "new_hosts": [],
+                "segments_with_running_basebackup": {1},
+                "is_pgrewind_running": [True, False, False],
+                "expected": [self._triplet('4|2|m|p|s|d|sdw2|sdw2|20000|/primary/gpseg2',
+                                           '8|2|p|m|s|u|sdw3|sdw3|21000|/mirror/gpseg2',
+                                           None)]
+            },
+            {
+                "name": "some_failed_segments_have_running_pgrewind",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "new_hosts": [],
+                "is_pgrewind_running": [True, False, True],
+                "expected": [self._triplet('3|1|m|p|s|d|sdw1|sdw1|20001|/primary/gpseg1',
+                                           '7|1|p|m|s|u|sdw2|sdw2|21001|/mirror/gpseg1',
+                                           None)]
+            },
+            # Moving to new hosts rewrites gpArray too; a skipped segment
+            # must keep its old host.
+            {
+                "name": "new_hosts_one_failed_segment_has_running_basebackup",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "new_hosts": ['new_1', 'new_2'],
+                "segments_with_running_basebackup": {0},
+                "expected": [self._triplet('3|1|m|p|s|d|sdw1|sdw1|20001|/primary/gpseg1',
+                                           '7|1|p|m|s|u|sdw2|sdw2|21001|/mirror/gpseg1',
+                                           '3|1|m|p|s|d|new_1|new_1|20001|/primary/gpseg1'),
+                             self._triplet('4|2|m|p|s|d|sdw2|sdw2|20000|/primary/gpseg2',
+                                           '8|2|p|m|s|u|sdw3|sdw3|21000|/mirror/gpseg2',
+                                           '4|2|m|p|s|d|new_2|new_2|20000|/primary/gpseg2')]
+            },
+            {
+                "name": "new_hosts_one_failed_segment_has_running_pgrewind",
+                "gparray": self.three_failedover_segs_gparray_str,
+                "new_hosts": ['new_1', 'new_2'],
+                "is_pgrewind_running": [False, True, False],
+                "expected": [self._triplet('2|0|m|p|s|d|sdw1|sdw1|20000|/primary/gpseg0',
+                                           '6|0|p|m|s|u|sdw2|sdw2|21000|/mirror/gpseg0',
+                                           '2|0|m|p|s|d|new_1|new_1|20000|/primary/gpseg0'),
+                             self._triplet('4|2|m|p|s|d|sdw2|sdw2|20000|/primary/gpseg2',
+                                           '8|2|p|m|s|u|sdw3|sdw3|21000|/mirror/gpseg2',
+                                           '4|2|m|p|s|d|new_2|new_2|20000|/primary/gpseg2')]
+            },
         ]
 
         self.run_pass_tests(tests, self.run_single_GpArray_test)
@@ -348,6 +511,68 @@ def test_RecoveryTripletsInPlaceAndNewHosts_getMirrorTriples_should_fail(self):
         ]
         self.run_fail_tests(tests, self.run_single_GpArray_test)
 
+    @patch('gppylib.db.dbconn.connect', side_effect=Exception())
+    def test_get_segments_with_running_basebackup_conn_exception(self, mock1):
+        with self.assertRaises(Exception) as ex:
+            get_segments_with_running_basebackup(15432)
+
+        self.assertTrue('Failed to query gp_stat_replication' in str(ex.exception))
+
+    @patch('gppylib.db.dbconn.connect', autospec=True)
+    @patch('gppylib.db.dbconn.query', side_effect=Exception())
+    def test_get_segments_with_running_basebackup_query_exception(self, mock1, mock2):
+        with self.assertRaises(Exception) as ex:
+            get_segments_with_running_basebackup(15432)
+
+        self.assertTrue('Failed to query gp_stat_replication:' in str(ex.exception))
+
+    @patch('gppylib.db.dbconn.connect', autospec=True)
+    @patch('gppylib.db.dbconn.query', return_value=FakeCursor(my_list=[]))
+    def test_get_segments_with_running_basebackup_no_basebackup(self, mock1, mock2):
+        segments_with_running_basebackup = get_segments_with_running_basebackup(15432)
+        self.assertEqual(list(segments_with_running_basebackup), [])
+        self.assertEqual([call("No basebackup running")], self.mock_logger.debug.call_args_list)
+
+    @patch('gppylib.db.dbconn.connect', autospec=True)
+    @patch('gppylib.db.dbconn.query', return_value=FakeCursor(my_list=[[1]]))
+    def test_get_segments_with_running_basebackup_has_basebackup(self, mock1, mock2):
+        segments_with_running_basebackup = get_segments_with_running_basebackup(15432)
+        self.assertEqual(list(segments_with_running_basebackup), [1])
+        self.assertEqual(0, self.mock_logger.debug.call_count)
+
+    # PGPORT need not be set, or set to the coordinator's port.
+    @patch.dict(os.environ, {'PGPORT': '5432'})
+    @patch('gppylib.db.dbconn.connect', autospec=True)
+    @patch('gppylib.db.dbconn.query', return_value=FakeCursor(my_list=[]))
+    def test_get_segments_with_running_basebackup_uses_coordinator_port(self, mock_query, mock_connect):
+        get_segments_with_running_basebackup(15432)
+        self.assertEqual(15432, mock_connect.call_args[0][0].pgport)
+
+    def test_triplets_check_basebackup_on_coordinator_port(self):
+        self.run_single_GpArray_test({
+            "gparray": self.all_up_gparray_str.replace('|mdw|mdw|5432|', '|mdw|mdw|15432|'),
+            "new_hosts": [],
+        })
+        gppylib.programs.clsRecoverSegment_triples.get_segments_with_running_basebackup.assert_called_once_with(15432)
+
+    @patch('gppylib.db.dbconn.connect', side_effect=Exception())
+    def test_is_pg_rewind_running_conn_exception(self, mock1):
+        with self.assertRaises(Exception) as ex:
+            is_pg_rewind_running("sdw1", 6001)
+        self.assertEqual('Failed to query pg_stat_activity for segment hostname: sdw1, port: 6001, error: ', str(ex.exception))
+
+    @patch('gppylib.db.dbconn.connect', autospec=True)
+    @patch('gppylib.db.dbconn.querySingleton', return_value=2)
+    def test_is_pg_rewind_running_returns_true_for_query_result_greater_than_one(self, mock1, mock2):
+        self.assertEqual(is_pg_rewind_running("sdw1", 6001), True)
+
+    @patch('gppylib.db.dbconn.connect', autospec=True)
+    @patch('gppylib.db.dbconn.querySingleton', return_value=0)
+    def test_is_pg_rewind_running_returns_false(self, mock1, mock2):
+        self.assertEqual(is_pg_rewind_running("sdw1", 6001), False)
+        self.assertEqual([call("Checking for running instances of pg_rewind with host sdw1 and port 6001 as source "
+                               "server")], self.mock_logger.debug.call_args_list)
+
     def run_pass_tests(self, tests, fn_to_test):
         for test in tests:
             with self.subTest(msg=test["name"]):
@@ -438,10 +663,13 @@ def __init__(self, arg):
                                   4|2|p|p|s|u|sdw2|sdw2|20000|/primary/gpseg2
                                   5|3|p|p|s|u|sdw2|sdw2|20001|/primary/gpseg3'''
 
-    def _run_single_FromGpArray_test(self, gparray_str, config_file, new_hosts, unreachable_hosts,
-                                              unreachable_existing_hosts=None):
+    def _run_single_FromGpArray_test(self, gparray_str, config_file, new_hosts, unreachable_hosts, is_pgrewind_running,
+                                     segments_with_running_basebackup, unreachable_existing_hosts=None):
         unreachable_hosts = unreachable_hosts if unreachable_hosts else []
         gppylib.programs.clsRecoverSegment_triples.get_unreachable_segment_hosts = Mock(return_value=unreachable_hosts)
+        gppylib.programs.clsRecoverSegment_triples.get_segments_with_running_basebackup = Mock(
+            return_value=segments_with_running_basebackup)
+        gppylib.programs.clsRecoverSegment_triples.is_pg_rewind_running = Mock(side_effect=is_pgrewind_running)
 
         initial_gparray = self.get_gp_array(gparray_str, unreachable_existing_hosts)
         mutated_gparray = self.get_gp_array(gparray_str, unreachable_existing_hosts)
@@ -540,7 +768,9 @@ def run_single_parser_test(test):
             "config": """sdw1|20000|/primary/gpseg0 sdw3|20001|/primary/gpseg5
                       sdw1|20001|/primary/gpseg1 sdw1|40001|/primary/gpseg_new
                       sdw3|20000|/primary/gpseg4
-                      sdw4|20000|/primary/gpseg6 sdw4|20000|/primary/gpseg6"""
+                      sdw4|20000|/primary/gpseg6 sdw4|20000|/primary/gpseg6
+                      sdw5|sdw5|20000|/primary/gpseg0 sdw3|10.0.34.5|20001|/primary/gpseg5
+                      sdw6|sdw6|20000|/primary/gpseg4"""
         },
         {
             "name": "6X_web_doc",
@@ -582,7 +812,7 @@ def test_parsing_should_pass(self):
                 """sdw1|20000|/mirror/gpseg0 sdw3|20001|/mirror/gpseg5
                    sdw1|20000 sdw3|20001|/mirror/gpseg5""",
             "expected":
-                "line 2 of file .*: expected 3 parts on failed segment group, obtained 2"
+                "line 2 of file .*: expected 3 or 4 parts on failed segment group, obtained 2"
         },
         {
             "name":
@@ -590,7 +820,7 @@ def test_parsing_should_pass(self):
             "config": """sdw1|20000|/mirror/gpseg0 sdw3|20001|/mirror/gpseg5
                          sdw2|50001|/data2/mirror/gpseg1 sdw4|50001""",
             "expected":
-                "line 2 of file .*: expected 3 parts on new segment group, obtained 2"
+                "line 2 of file .*: expected equal parts, either 3 or 4 on both segment group, obtained 3 on group1 and 2 on group2"
         },
         {
             "name":
@@ -697,7 +927,25 @@ def test_parsing_should_pass(self):
             "name": "new_port_invalid",
             "config": """sdw2|50001|/data2/mirror/gpseg1 sdw4|new_invalid_port|relative/new/mirror/gpseg1""",
             "expected": "Invalid port on line 1"
-        }
+        },
+        {
+            "name":
+                "invalid_parts_present_in_group_1",
+            "config":
+                """sdw1|10.0.34.2|20000|/primary/gpseg0 sdw3|10.0.34.5|20001|/primary/gpseg5
+                   sdw1|20000|/primary/gpseg0 sdw3|10.0.34.5|20001|/primary/gpseg5""",
+            "expected":
+                "line 2 of file .*: expected equal parts, either 3 or 4 on both segment group, obtained 3 on group1 and 4 on group2"
+        },
+        {
+            "name":
+                "invalid_parts_present_in_group_2",
+            "config":
+                """sdw1|10.0.34.2|20000|/primary/gpseg0 sdw3|10.0.34.5|20001|/primary/gpseg5
+                   sdw1|10.0.34.2|20000|/primary/gpseg0 sdw3|20001|/primary/gpseg5""",
+            "expected":
+                "line 2 of file .*: expected equal parts, either 3 or 4 on both segment group, obtained 4 on group1 and 3 on group2"
+        },
     ]
 
     def test_parsing_should_fail(self):
@@ -717,21 +965,30 @@ def _get_expected_config(config_str):
         for line in config_str.splitlines():
             lineno += 1
             groups = line.split()
+            parts = groups[0].split('|')
+            hostname, address, port, datadir, hostname_check_required = extract_recovery_config_info(parts)
 
-            address, port, datadir = groups[0].split('|')
             row = {
+                'failedHostname': hostname,
                 'failedAddress': address,
                 'failedPort': port,
                 'failedDataDirectory': datadir,
-                'lineno': lineno
+                'lineno': lineno,
+                'hostname_check_required': hostname_check_required
             }
 
             if len(groups) > 1:
-                address, port, datadir = groups[1].split('|')
+                parts2 = groups[1].split('|')
+                if len(parts2) == 4:
+                    hostname2, address2, port2, datadir2 = parts2
+                else:
+                    address2, port2, datadir2 = parts2
+                    hostname2 = address2
                 row.update({
-                    'newAddress': address,
-                    'newPort': port,
-                    'newDataDirectory': datadir
+                    'newHostname': hostname2,
+                    'newAddress': address2,
+                    'newPort': port2,
+                    'newDataDirectory': datadir2
                 })
 
             rows.append(row)
diff --git a/gpMgmt/bin/gppylib/recoveryinfo.py b/gpMgmt/bin/gppylib/recoveryinfo.py
index f13ea81394b..b4530bc2194 100644
--- a/gpMgmt/bin/gppylib/recoveryinfo.py
+++ b/gpMgmt/bin/gppylib/recoveryinfo.py
@@ -1,8 +1,10 @@
 from collections import defaultdict
 import datetime
 import json
+import shlex
 
 from gppylib import gplog
+from gppylib.commands.base import Command, REMOTE
 
 
 class RecoveryInfo(object):
@@ -169,8 +171,36 @@ def print_setup_recovery_errors(self):
                     self._logger.error(setup_recovery_error_pattern.format(hostname, error.port, error.error_msg))
         self._print_invalid_errors()
 
+    # Cherry-picked from Greenplum Database (greenplum-db/gpdb-archive @ 482967c1b4),
+    # gpMgmt/bin/gppylib/recoveryinfo.py, with the line it filters out
+    # changed: pg_rewind reports its own failure differently since
+    # PostgreSQL 13. See the comment in the body.
+    def get_error_from_logfile(self, progress_file, hostname):
+        """
+        Traverse the error file to fetch last registered occurrence of 'error, panic or fatal'.
+        pg_rewind logs a line of its own after every failed attempt, which is
+        never the reason for the failure -- the reason is whatever the backend
+        wrote just above it -- so that line is excluded. Upstream excludes
+        PostgreSQL 12's wording of it, "fatal: postgres single-user mode of
+        target instance failed for command"; since 13 pg_rewind writes
+        "error: postgres single-user mode in target cluster failed" followed by
+        a separate "detail: Command was: ...", and only the first of those two
+        carries a word this grep looks for.
+        To ensure grep doesn't return a non-zero exit code, it is ORed with true.
+        """
+        cmdStr = r'set -o pipefail; cat {} | (grep -i "ERROR\|PANIC\|FATAL" | grep -v "error: postgres single-user mode in target cluster failed" || true) | tail -1'.format(
+            shlex.quote(progress_file))
+        cmd = Command(name="Parse logfile for errors on a remote host", cmdStr=cmdStr, ctxt=REMOTE, remoteHost=hostname)
+        cmd.run()
+
+        if cmd.get_results().rc != 0:
+            self._logger.debug("Failed while parsing the logfile %s" % progress_file)
+            return None
+
+        return cmd.get_results().stdout.strip()
+
     def print_bb_rewind_and_start_errors(self):
-        bb_rewind_error_pattern = " hostname: {}; port: {}; logfile: {}; recoverytype: {}"
+        bb_rewind_error_pattern = " hostname: {}; port: {}; logfile: {}; recoverytype: {}; error: {}"
         if len(self._bb_errors) > 0 or len(self._rewind_errors) > 0:
             self._logger.info("----------------------------------------------------------")
             if len(self._rewind_errors) > 0:
@@ -181,11 +211,15 @@ def print_bb_rewind_and_start_errors(self):
             for hostname, errors in self._rewind_errors.items():
                 for error in errors:
                     self._logger.info(bb_rewind_error_pattern.format(hostname, error.port, error.progress_file,
-                                                                     error.error_type))
+                                                                     error.error_type,
+                                                                     self.get_error_from_logfile(error.progress_file,
+                                                                                                 hostname)))
             for hostname, errors in self._bb_errors.items():
                 for error in errors:
                     self._logger.info(bb_rewind_error_pattern.format(hostname, error.port, error.progress_file,
-                                                                 error.error_type))
+                                                                 error.error_type,
+                                                                 self.get_error_from_logfile(error.progress_file,
+                                                                                             hostname)))
 
         start_error_pattern = " hostname: {}; port: {}; datadir: {}"
         if len(self._start_errors) > 0:
diff --git a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py
index 983b5f25a9a..afc7438496c 100644
--- a/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py
+++ b/gpMgmt/bin/gppylib/test/unit/test_unit_gpsegrecovery.py
@@ -1,7 +1,10 @@
 from contextlib import redirect_stderr
 from mock import call, Mock, patch, ANY
 import io
+import os
+import shutil
 import sys
+import tempfile
 
 from .gp_unittest import GpTestCase
 import gpsegrecovery
@@ -18,11 +21,13 @@ def setUp(self):
         self.mock_logger = Mock()
         self.apply_patches([
             patch('gpsegrecovery.start_segment', return_value=Mock()),
+            patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()),
             patch('gppylib.commands.pg.PgRewind.__init__', return_value=None),
             patch('gppylib.commands.pg.PgRewind.run')
         ])
         self.mock_pgrewind_run = self.get_mock_from_apply_patch('run')
         self.mock_pgrewind_init = self.get_mock_from_apply_patch('__init__')
+        self.mock_modify_conf = self.get_mock_from_apply_patch('ModifyConfSetting')
 
         p = Segment.initFromString("1|0|p|p|s|u|sdw1|sdw1|40000|/data/primary0")
         m = Segment.initFromString("2|0|m|m|s|u|sdw2|sdw2|50000|/data/mirror0")
@@ -56,8 +61,13 @@ def test_incremental_run_passes(self):
         self.assertEqual(1, self.mock_pgrewind_run.call_count)
         self.assertEqual(call(validateAfter=True), self.mock_pgrewind_run.call_args)
         logger_call_args = [call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file'),
-                            call('Successfully ran pg_rewind for dbid: 2')]
+                            call('Successfully ran pg_rewind for dbid: 2'),
+                            call('Updating /data/mirror0/postgresql.conf')]
         self.assertEqual(logger_call_args, self.mock_logger.info.call_args_list)
+        self.mock_modify_conf.assert_called_once_with('Updating /data/mirror0/postgresql.conf',
+                                                      '/data/mirror0/postgresql.conf', 'port', 50000,
+                                                      optType='number')
+        self.mock_modify_conf.return_value.run.assert_called_once_with(validateAfter=True)
         gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era)
 
     def test_incremental_run_exception(self):
@@ -71,6 +81,7 @@ def test_incremental_run_exception(self):
         self.assertEqual(call(validateAfter=True), self.mock_pgrewind_run.call_args)
         self.assertEqual([call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file')],
                          self.mock_logger.info.call_args_list)
+        self.assertEqual(0, self.mock_modify_conf.call_count)
         self.assertEqual(0, gpsegrecovery.start_segment.call_count)
         self._assert_cmd_failed('{"error_type": "incremental", "error_msg": "pg_rewind failed", "dbid": 2, ' \
                                 '"datadir": "/data/mirror0", "port": 50000, "progress_file": "/tmp/test_progress_file"}')
@@ -92,7 +103,8 @@ def test_incremental_start_segment_exception(self):
         self.assertEqual(1, self.mock_pgrewind_init.call_count)
         self.assertEqual(1, self.mock_pgrewind_run.call_count)
         logger_call_args = [call('Running pg_rewind with progress output temporarily in /tmp/test_progress_file'),
-                            call('Successfully ran pg_rewind for dbid: 2')]
+                            call('Successfully ran pg_rewind for dbid: 2'),
+                            call('Updating /data/mirror0/postgresql.conf')]
         self.assertEqual(logger_call_args, self.mock_logger.info.call_args_list)
         gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era)
         self._assert_cmd_failed('{"error_type": "start", "error_msg": "pg_ctl start failed", "dbid": 2, ' \
@@ -107,11 +119,13 @@ def setUp(self):
         self.mock_logger = Mock(spec=['log', 'info', 'debug', 'error', 'warn', 'exception'])
         self.apply_patches([
             patch('gpsegrecovery.start_segment', return_value=Mock()),
+            patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()),
             patch('gpsegrecovery.PgBaseBackup.__init__', return_value=None),
             patch('gpsegrecovery.PgBaseBackup.run')
         ])
         self.mock_pgbasebackup_run = self.get_mock_from_apply_patch('run')
         self.mock_pgbasebackup_init = self.get_mock_from_apply_patch('__init__')
+        self.mock_modify_conf = self.get_mock_from_apply_patch('ModifyConfSetting')
 
         p = Segment.initFromString("1|0|p|p|s|u|sdw1|sdw1|40000|/data/primary0")
         m = Segment.initFromString("2|0|m|m|s|u|sdw2|sdw2|50000|/data/mirror0")
@@ -135,8 +149,13 @@ def _assert_basebackup_runs(self, expected_init_args):
         self.assertEqual(1, self.mock_pgbasebackup_run.call_count)
         self.assertEqual(call(validateAfter=True), self.mock_pgbasebackup_run.call_args)
         expected_logger_info_args = [call('Running pg_basebackup with progress output temporarily in /tmp/test_progress_file'),
-                                     call("Successfully ran pg_basebackup for dbid: 2")]
+                                     call("Successfully ran pg_basebackup for dbid: 2"),
+                                     call('Updating /data/mirror0/postgresql.conf')]
         self.assertEqual(expected_logger_info_args, self.mock_logger.info.call_args_list)
+        self.mock_modify_conf.assert_called_once_with('Updating /data/mirror0/postgresql.conf',
+                                                      '/data/mirror0/postgresql.conf', 'port', 50000,
+                                                      optType='number')
+        self.mock_modify_conf.return_value.run.assert_called_once_with(validateAfter=True)
         gpsegrecovery.start_segment.assert_called_once_with(self.seg_recovery_info, self.mock_logger, self.era)
 
     def _assert_cmd_passed(self):
@@ -277,6 +296,7 @@ def setUp(self):
         self.apply_patches([
             patch('gpsegrecovery.SegmentStart.__init__', return_value=None),
             patch('gpsegrecovery.SegmentStart.run'),
+            patch('gpsegrecovery.ModifyConfSetting', return_value=Mock()),
         ])
         self.mock_segment_start_init = self.get_mock_from_apply_patch('__init__')
 
@@ -396,3 +416,69 @@ def test_start_segment_passes(self, mock_run, mock_init):
         mock_init.assert_called_once()
         self.assertEqual(1, self.mock_logger.info.call_count)
         mock_run.assert_called_once_with(validateAfter=True)
+
+
+class UpdatePortInConfTestCase(GpTestCase):
+    """
+    Runs the real rewrite, not a mock of it: the rewrite reports success
+    whether or not it matched a line, so only the file shows whether the port
+    was changed.
+    """
+    def setUp(self):
+        self.datadir = tempfile.mkdtemp()
+        self.recovery_info = RecoveryInfo(self.datadir, 50000, 2, 'sdw1', 40000, False, '/tmp/test_progress_file')
+
+    def tearDown(self):
+        shutil.rmtree(self.datadir)
+        super(UpdatePortInConfTestCase, self).tearDown()
+
+    def _rewrite(self, line):
+        conf = os.path.join(self.datadir, 'postgresql.conf')
+        with open(conf, 'w') as f:
+            f.write(line + '\n')
+        gpsegrecovery.update_port_in_conf(self.recovery_info, Mock())
+        with open(conf) as f:
+            return f.read().rstrip('\n')
+
+    def test_every_form_postgresql_conf_accepts_is_rewritten(self):
+        forms = ['port=40000', 'port = 40000', ' port = 40000', 'port\t=40000', '\tport=40000',
+                 'port 40000', "port = '40000'", 'PORT = 40000', 'Port=40000']
+        for form in forms:
+            with self.subTest(form=form):
+                self.assertEqual('port=50000', self._rewrite(form))
+
+    def test_trailing_comment_is_kept(self):
+        self.assertEqual('port=50000 # sets the listener port',
+                         self._rewrite('port = 40000 # sets the listener port'))
+
+    def test_commented_out_and_other_settings_are_left_alone(self):
+        for line in ['#port = 5432', '#PORT = 5432', 'port_range = 40000', 'PORT_RANGE = 40000']:
+            with self.subTest(line=line):
+                self.assertEqual(line, self._rewrite(line))
+
+    def test_alternative_numeric_values_are_replaced_completely(self):
+        values = ['+40000', "'+40000'", '0116100', "'0116100'",
+                  '0x9c40', "'0x9c40'", '40000.0', "'40000.0'", "'4e4'"]
+        for value in values:
+            with self.subTest(value=value):
+                self.assertEqual('port=50000', self._rewrite('port = ' + value))
+                self.assertEqual('port=50000 # source port',
+                                 self._rewrite('PORT ' + value + ' # source port'))
+
+    def test_last_port_assignment_is_also_rewritten(self):
+        self.assertEqual('port=50000\nport=50000 # override',
+                         self._rewrite("port=40000\nPORT = '0x9c40' # override"))
+
+    def test_rewrite_preserves_adjacent_lines(self):
+        self.assertEqual("# source config\n\nport=50000\nport_range = 40000",
+                         self._rewrite("# source config\n\nport = '4e4'\nport_range = 40000"))
+
+    def test_crlf_lines_are_rewritten_and_line_endings_preserved(self):
+        conf = os.path.join(self.datadir, 'postgresql.conf')
+        with open(conf, 'wb') as f:
+            f.write(b"# source config\r\nport = 40000\r\nPORT = '40000' \t\r\n"
+                    b"port = 40000 # override\r\nport_range = 40000\r\n")
+        gpsegrecovery.update_port_in_conf(self.recovery_info, Mock())
+        with open(conf, 'rb') as f:
+            self.assertEqual(b"# source config\r\nport=50000\r\nport=50000 \t\r\n"
+                             b"port=50000 # override\r\nport_range = 40000\r\n", f.read())
diff --git a/gpMgmt/bin/gppylib/utils.py b/gpMgmt/bin/gppylib/utils.py
index 42fca86592e..bc5af671fa5 100644
--- a/gpMgmt/bin/gppylib/utils.py
+++ b/gpMgmt/bin/gppylib/utils.py
@@ -612,3 +612,45 @@ def formatInsertValuesList(row, starelid, inclHLL):
         rowVals.append('\t{0}::{1}'.format(val, typ))
 
     return rowVals
+
+
+def _resolve_all_addresses(name):
+    """
+    Every address a name resolves to, across address families.
+
+    Not gethostbyaddr(): it answers for one family only, so on a dual-stack
+    host "localhost" can come back as ::1 while an address that names the same
+    machine comes back as 127.0.0.1. Comparing those two lists says the names
+    are different hosts when they are not.
+    """
+    return {info[4][0] for info in socket.getaddrinfo(name, None)}
+
+
+# cherry-pick from Greenplum Database
+# greenplum-db/gpdb-archive @ 482967c1b4, gpMgmt/bin/gppylib/utils.py
+def validateHostnameAddress(hostname, address):
+    """
+    validateHostnameAddress : validates that given hostname and address are for the same host
+    Address can be hostname/alias also. Resolves hostname and address both to get the associated addresses.
+
+    @param hostname Name of the host to be validated
+    @param address Address of the host to be validated. Can be hostname/alias also
+    @return if the address and hostname are of the same host
+    """
+    try:
+        resolved_address_list = _resolve_all_addresses(hostname)
+        resolved_address_list_2 = _resolve_all_addresses(address)
+    except Exception as e:
+        # This means given hostname or address is not reachable
+        logger.warning(
+            "Could not resolve hostname:{0}."
+                .format(hostname))
+        return False
+
+    # Resolved address and hostname should have at least one IP address common if they are of same host
+    if not bool(resolved_address_list.intersection(resolved_address_list_2)):
+        logger.warning(
+            "Given address:{0} not present in resolved hostname:{1} address list we got:{2}".format(
+                address, hostname, sorted(resolved_address_list)))
+        return False
+    return True
diff --git a/gpMgmt/sbin/gpsegrecovery.py b/gpMgmt/sbin/gpsegrecovery.py
index 811a73ccbb5..a1a6891ee35 100644
--- a/gpMgmt/sbin/gpsegrecovery.py
+++ b/gpMgmt/sbin/gpsegrecovery.py
@@ -7,7 +7,7 @@
 from gppylib.commands.pg import PgBaseBackup, PgRewind
 from recovery_base import RecoveryBase, set_recovery_cmd_results
 from gppylib.commands.base import Command
-from gppylib.commands.gp import SegmentStart
+from gppylib.commands.gp import SegmentStart, ModifyConfSetting
 from gppylib.gparray import Segment
 from gppylib.commands.unix import terminate_proc_tree
 
@@ -63,6 +63,10 @@ def run(self):
         self.error_type = RecoveryErrorType.DEFAULT_ERROR
         self.logger.info("Successfully ran pg_basebackup for dbid: {}".format(
             self.recovery_info.target_segment_dbid))
+
+        # Updating port number on conf after recovery
+        update_port_in_conf(self.recovery_info, self.logger)
+
         self.error_type = RecoveryErrorType.START_ERROR
         start_segment(self.recovery_info, self.logger, self.era)
 
@@ -87,10 +91,27 @@ def run(self):
         cmd.run(validateAfter=True)
         self.logger.info("Successfully ran pg_rewind for dbid: {}".format(self.recovery_info.target_segment_dbid))
 
+        # Updating port number on conf after recovery
+        update_port_in_conf(self.recovery_info, self.logger)
+
         self.error_type = RecoveryErrorType.START_ERROR
         start_segment(self.recovery_info, self.logger, self.era)
 
 
+def update_port_in_conf(recovery_info, logger):
+    """
+    pg_basebackup and pg_rewind both copy the source segment's
+    postgresql.conf, which carries the *source* port. Point it back at this
+    segment before it is started, or the mirror comes up on its primary's
+    port.
+    """
+    logger.info("Updating %s/postgresql.conf" % recovery_info.target_datadir)
+    modifyConfCmd = ModifyConfSetting('Updating %s/postgresql.conf' % recovery_info.target_datadir,
+                                      "{}/{}".format(recovery_info.target_datadir, 'postgresql.conf'),
+                                      'port', recovery_info.target_port, optType='number')
+    modifyConfCmd.run(validateAfter=True)
+
+
 def start_segment(recovery_info, logger, era):
     seg = Segment(None, None, None, None, None, None, None, None,
                   recovery_info.target_port, recovery_info.target_datadir)
diff --git a/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature b/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature
index b63b279caae..9817183b579 100644
--- a/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature
+++ b/gpMgmt/test/behave/mgmt_utils/gpaddmirrors.feature
@@ -38,8 +38,6 @@ Feature: Tests for gpaddmirrors
         | -B 2 -b 1 |  2                  |  1              |
         | -B 1 -b 2 |  1                  |  2              |
 
-    # needs_fix: a recovered mirror keeps its primary's port in postgresql.conf (#1639 a3bfe47d5c3)
-    @needs_fix
     Scenario: gpaddmirrors fails for recovery setup errors
         Given the cluster is generated with "3" primaries only
         When gpaddmirrors adds 3 mirrors with one mirror's datadir not empty
@@ -155,8 +153,6 @@ Feature: Tests for gpaddmirrors
             | all          | None                   | 0,1,2            |
 
 
-    # needs_fix: a recovered mirror keeps its primary's port in postgresql.conf (#1639 a3bfe47d5c3)
-    @needs_fix
     Scenario: gpaddmirrors can add mirrors and display progress in gpstate
         Given the cluster is generated with "3" primaries only
         And all files in gpAdminLogs directory are deleted
diff --git a/gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature b/gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature
index 5f527af6035..06aa918bb8f 100644
--- a/gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature
+++ b/gpMgmt/test/behave/mgmt_utils/gpmovemirrors.feature
@@ -22,8 +22,6 @@ Feature: Tests for gpmovemirrors
         When the user runs gpmovemirrors with additional args "--invalid-option"
         Then gpmovemirrors should return a return code of 2
 
-    # needs_fix: gprecoverseg/gppylib fixes (#1639 a3bfe47d5c3 / 51955ac3d8f / 7056e02e306)
-    @needs_fix
     Scenario: gpmovemirrors can change the location of mirrors within a single host
         Given a standard local demo cluster is created
         And a gpmovemirrors directory under '/tmp/gpmovemirrors' with mode '0700' is created
@@ -39,8 +37,6 @@ Feature: Tests for gpmovemirrors
         And check segment conf: postgresql.conf
         And verify that mirrors are recognized after a restart
 
-    # needs_fix: gprecoverseg/gppylib fixes (#1639 a3bfe47d5c3 / 51955ac3d8f / 7056e02e306)
-    @needs_fix
     Scenario: gpmovemirrors can change the port of mirrors within a single host
         Given a standard local demo cluster is created
         And a gpmovemirrors directory under '/tmp/gpmovemirrors' with mode '0700' is created
@@ -117,8 +113,6 @@ Feature: Tests for gpmovemirrors
                 assert that gp_seg_config wasn't updated
         """
     @skip_cleanup
-    # needs_fix: gprecoverseg/gppylib fixes (#1639 a3bfe47d5c3 / 51955ac3d8f / 7056e02e306)
-    @needs_fix
     Scenario Outline: user can <correction> if <failed_count> mirrors failed to move initially
         Given the database is running
         And all the segments are running
@@ -163,8 +157,6 @@ Feature: Tests for gpmovemirrors
         | run gprecoverseg          | all          | None               | 0,1,2          | running in place full recovery for all failed contents |
 
     @skip_cleanup
-    # needs_fix: gprecoverseg/gppylib fixes (#1639 a3bfe47d5c3 / 51955ac3d8f / 7056e02e306)
-    @needs_fix
     Scenario: gpmovemirrors can move mirrors even if start fails for some mirrors
         Given the database is running
         And all the segments are running
@@ -251,8 +243,6 @@ Feature: Tests for gpmovemirrors
   @demo_cluster
   @concourse_cluster
   @skip_cleanup
-  # needs_fix: skip segments that already have a recovery running (#1639 7056e02e306)
-  @needs_fix
   Scenario: gpmovemirrors gives warning if pg_basebackup is already running for one of the mirrors to be moved
     Given the database is running
     And all the segments are running
@@ -292,8 +282,6 @@ Feature: Tests for gpmovemirrors
   @demo_cluster
   @concourse_cluster
   @skip_cleanup
-  # needs_fix: skip segments that already have a recovery running (#1639 7056e02e306)
-  @needs_fix
   Scenario: gpmovemirrors gives warning if pg_basebackup is already running for some of the mirrors to be moved
     Given the database is running
     And all the segments are running
@@ -334,8 +322,6 @@ Feature: Tests for gpmovemirrors
   @demo_cluster
   @concourse_cluster
   @skip_cleanup
-  # needs_fix: skip segments that already have a recovery running (#1639 7056e02e306)
-  @needs_fix
   Scenario: gpmovemirrors gives warning if pg_basebackup is already running for all mirrors to be moved
     Given the database is running
     And all the segments are running
diff --git a/gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature b/gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature
index 68d7be4b053..151d9f2f0dc 100644
--- a/gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature
+++ b/gpMgmt/test/behave/mgmt_utils/gprecoverseg.feature
@@ -1,7 +1,4 @@
-# needs_fix: the gprecoverseg fixes from #1639 (a3bfe47d5c3, 879b5638026,
-# 7056e02e306, 0aa26662c2c, 51955ac3d8f); without them a scenario hangs and
-# the job reaches its timeout.
-@gprecoverseg @needs_fix
+@gprecoverseg
 Feature: gprecoverseg tests
 
     @differential
@@ -816,8 +813,6 @@ Feature: gprecoverseg tests
 
   @demo_cluster
   @concourse_cluster
-  # needs_fix: skip segments that already have a recovery running (#1639 7056e02e306)
-  @needs_fix
   Scenario: gprecoverseg gives warning if pg_basebackup already running for one of the failed segments
     Given the database is running
     And all the segments are running
@@ -850,8 +845,6 @@ Feature: gprecoverseg tests
 
   @demo_cluster
   @concourse_cluster
-  # needs_fix: skip segments that already have a recovery running (#1639 7056e02e306)
-  @needs_fix
   Scenario: gprecoverseg gives warning if pg_basebackup already running for some of the failed segments
     Given the database is running
     And all the segments are running
@@ -886,8 +879,6 @@ Feature: gprecoverseg tests
 
   @demo_cluster
   @concourse_cluster
-  # needs_fix: skip segments that already have a recovery running (#1639 7056e02e306)
-  @needs_fix
   Scenario: gprecoverseg -aF gives warning if pg_basebackup already running for all of the failed segments
     Given the database is running
     And all the segments are running
@@ -921,8 +912,6 @@ Feature: gprecoverseg tests
 
   @demo_cluster
   @concourse_cluster
-  # needs_fix: skip segments that already have a recovery running (#1639 7056e02e306)
-  @needs_fix
   Scenario: gprecoverseg -i gives warning if pg_basebackup already running for all failed segments
     Given the database is running
     And all the segments are running
@@ -1787,8 +1776,6 @@ Feature: gprecoverseg tests
 
 
   @concourse_cluster
-    # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-    @needs_fix
     Scenario: gprecoverseg recovery to new host populates hostname and address from the config file correctly
         Given the database is running
           And all the segments are running
@@ -1806,8 +1793,6 @@ Feature: gprecoverseg tests
           And the segments are synchronized
 
     @concourse_cluster
-    # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-    @needs_fix
     Scenario: gprecoverseg recovery to same host (full inplace) populates hostname and address from the config file correctly
         Given the database is running
           And all the segments are running
@@ -1826,8 +1811,6 @@ Feature: gprecoverseg tests
 
 
   @concourse_cluster
-    # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-    @needs_fix
     Scenario: gprecoverseg recovery with invalid format with hostname in config file
         Given the database is running
           And all the segments are running
@@ -1847,8 +1830,6 @@ Feature: gprecoverseg tests
 
 
   @concourse_cluster
-    # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-    @needs_fix
     Scenario: gprecoverseg incremental recovery populates hostname and address from the config file correctly
         Given the database is running
           And all the segments are running
@@ -1866,8 +1847,6 @@ Feature: gprecoverseg tests
           And the segments are synchronized
 
     @concourse_cluster
-    # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-    @needs_fix
     Scenario: gprecoverseg recovery with and without hostname parameter in config file
         Given the database is running
           And all the segments are running
@@ -1882,8 +1861,6 @@ Feature: gprecoverseg tests
           And the segments are synchronized
 
     @concourse_cluster
-    # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-    @needs_fix
     Scenario: gprecoverseg throws warning and skips recovery if provided hostname and address can not be resolved to same host
         Given the database is running
           And all the segments are running
@@ -1902,8 +1879,6 @@ Feature: gprecoverseg tests
           And the segments are synchronized
 
     @concourse_cluster
-    # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-    @needs_fix
     Scenario: gprecoverseg incremental recovery fails if config file contains wrong hostname of failed segment
         Given the database is running
           And all the segments are running
@@ -1921,8 +1896,6 @@ Feature: gprecoverseg tests
           And gprecoverseg should return a return code of 0
           And the cluster is rebalanced
 
-  # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-  @needs_fix
   @demo_cluster
   Scenario: gprecoverseg recovers segment when config file contains hostname on demo cluster
     Given the database is running
@@ -1941,8 +1914,6 @@ Feature: gprecoverseg tests
     And the cluster configuration has no segments where "content=0 and status='d'"
     Then the cluster is rebalanced
 
-  # needs_fix: hostname in the -i recovery config file (#1639 0aa26662c2c)
-  @needs_fix
   @demo_cluster
   Scenario: gprecoverseg skips recovery when config file contains invalid hostname on demo cluster
     Given the database is running