diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst
index 1b548057..4728dce8 100644
--- a/docs/SETTINGS.rst
+++ b/docs/SETTINGS.rst
@@ -70,6 +70,8 @@ PostgreSQL
 -  **recovery\_conf**: additional configuration settings written to recovery.conf when configuring follower.
 -  **parameters**: list of configuration settings for Postgres. Many of these are required for replication to work.
 -  **pg\_ctl\_timeout**: How long should pg_ctl wait when doing ``start``, ``stop`` or ``restart``. Default value is 60 seconds.
+-  **use\_pg\_rewind**: try to use pg\_rewind on the former leader when it joins cluster as a replica.
+-  **remove\_data\_directory\_on\_rewind\_failure**: If this option is enabled, Patroni will remove postgres data directory and recreate replica. Otherwise it will try to follow the new leader. Default value is **false**.
 -  **replica\_method** for each create_replica_method other than basebackup, you would add a configuration section of the same name. At a minimum, this should include "command" with a full path to the actual script to be executed.  Other configuration parameters will be passed along to the script in the form "parameter=value".
 
 REST API
diff --git a/patroni/postgresql.py b/patroni/postgresql.py
index 45ed5c92..97482b18 100644
--- a/patroni/postgresql.py
+++ b/patroni/postgresql.py
@@ -87,7 +87,6 @@ class Postgresql(object):
         self._replication = config['authentication']['replication']
         self.resolve_connection_addresses()
 
-        self._use_pg_rewind = config.get('use_pg_rewind', False)
         self._need_rewind = False
         self._use_slots = config.get('use_slots', True)
         self._version_file = os.path.join(self._data_dir, 'PG_VERSION')
@@ -238,7 +237,7 @@ class Postgresql(object):
             we have either wal_log_hints or checksums turned on
         """
         # low-hanging fruit: check if pg_rewind configuration is there
-        if not (self._use_pg_rewind and all(self._superuser.get(n) for n in ('username', 'password'))):
+        if not (self.config.get('use_pg_rewind') and all(self._superuser.get(n) for n in ('username', 'password'))):
             return False
 
         cmd = ['pg_rewind', '--help']
@@ -730,22 +729,25 @@ class Postgresql(object):
 
         change_role = self.role == 'master'
 
-        if leader and leader.name == self.name:
-            self._need_rewind = False
-            member = None
-            if self.is_running():
-                return
-        else:
-            self._need_rewind = self._need_rewind or change_role and self.can_rewind
+        if change_role:
+            if leader:
+                if leader.name == self.name:
+                    self._need_rewind = False
+                    member = None
+                    if self.is_running():
+                        return
+                else:
+                    self._need_rewind = bool(leader.conn_url) and self.can_rewind
+            else:
+                self._need_rewind = False
+                member = None
 
         if self._need_rewind:
             logger.info("set the rewind flag after demote")
 
-            if self.is_running():
-                stopped = self.stop()
-                self.set_role('unknown')
-                if not stopped:
-                    return logger.warning('Can not run pg_rewind because postgres is still running')
+            self.set_role('unknown')
+            if self.is_running() and not self.stop():
+                return logger.warning('Can not run pg_rewind because postgres is still running')
 
             if not (leader and leader.conn_url):
                 return logger.info('Leader unknown, can not rewind')
@@ -776,7 +778,7 @@ class Postgresql(object):
             opts.update({'archive_mode': 'on', 'archive_command': 'false'})
             self.single_user_mode(options=opts)
 
-            if self.rewind(r):
+            if self.rewind(r) or not self.config.get('remove_data_directory_on_rewind_failure', False):
                 self.write_recovery_conf(member)
                 ret = self.start()
             else:
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index f744e401..93144aa3 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -171,6 +171,7 @@ class TestPostgresql(unittest.TestCase):
                              'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432',
                              'authentication': {'superuser': {'username': 'test', 'password': 'test'},
                                                 'replication': {'username': 'replicator', 'password': 'rep-pass'}},
+                             'remove_data_directory_on_rewind_failure': True,
                              'use_pg_rewind': True, 'pg_ctl_timeout': 'bla',
                              'parameters': self._PARAMETERS,
                              'recovery_conf': {'foo': 'bar'},
@@ -281,14 +282,14 @@ class TestPostgresql(unittest.TestCase):
 
     @patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
     def test_can_rewind(self):
-        with mock.patch('subprocess.call', MagicMock(return_value=1)):
+        with patch('subprocess.call', MagicMock(return_value=1)):
             self.assertFalse(self.p.can_rewind)
-        with mock.patch('subprocess.call', side_effect=OSError):
+        with patch('subprocess.call', side_effect=OSError):
             self.assertFalse(self.p.can_rewind)
-        tmp = self.p.controldata
-        self.p.controldata = lambda: {'wal_log_hints setting': 'on'}
-        self.assertTrue(self.p.can_rewind)
-        self.p.controldata = tmp
+        with patch.object(Postgresql, 'controldata', Mock(return_value={'wal_log_hints setting': 'on'})):
+            self.assertTrue(self.p.can_rewind)
+        self.p.config['use_pg_rewind'] = False
+        self.assertFalse(self.p.can_rewind)
 
     @patch('time.sleep', Mock())
     def test_create_replica(self):