logical replication

Author: jeeminso

tests/bwc/test_rolling_upgrade.py

@@ -1,3 +1,4 @@
+import time
 import unittest
 from crate.client import connect
 from crate.client.exceptions import ProgrammingError
@@ -88,6 +89,9 @@ def _test_rolling_upgrade(self, path, nodes):
         }
         cluster = self._new_cluster(path.from_version, nodes, settings=settings)
         cluster.start()
+        replica_cluster = self._new_cluster(path.from_version, 1, settings=settings)
+        replica_cluster.start()
+        cluster.node()
         with connect(cluster.node().http_url, error_trace=True) as conn:
             c = conn.cursor()
             c.execute("create user arthur with (password = 'secret')")
@@ -133,6 +137,15 @@ def _test_rolling_upgrade(self, path, nodes):
             # Add the shards of the new partition primaries
             expected_active_shards += shards
 
+            c.execute("create table doc.x (a int)")
+            c.execute("create publication p for table doc.x")
+            with connect(replica_cluster.node().http_url, error_trace=True) as replica_conn:
+                rc = replica_conn.cursor()
+                rc.execute(f"create subscription rs connection 'crate://{cluster.node().http_url}?user=crate&sslmode=sniff' publication p")
+                rc.execute("create table doc.rx (a int)")
+                rc.execute("create publication rp for table doc.rx")
+            c.execute(f"create subscription s connection 'crate://{replica_cluster.node().http_url}?user=crate&sslmode=sniff' publication rp")
+
         for idx, node in enumerate(cluster):
             # Enforce an old version node be a handler to make sure that an upgraded node can serve 'select *' from an old version node.
             # Otherwise upgraded node simply requests N-1 columns from old version with N columns and it always works.
@@ -236,6 +249,25 @@ def _test_rolling_upgrade(self, path, nodes):
                 # Add the shards of the new partition primaries
                 expected_active_shards += shards
 
+                with connect(replica_cluster.node().http_url, error_trace=True) as replica_conn:
+                    rc = replica_conn.cursor()
+                    # Ensure publishing to remote cluster works
+                    rc.execute("select count(*) from doc.x")
+                    count = rc.fetchall()[0][0]
+                    c.execute("insert into doc.x values (1)")
+                    rc.execute("refresh table doc.x")
+                    time.sleep(1) # replication delay...
+                    rc.execute("select count(*) from doc.x")
+                    self.assertEqual(rc.fetchall()[0][0], count + 1)
+
+                    # Ensure subscription from remote cluster works
+                    c.execute("select count(*) from doc.rx")
+                    count = c.fetchall()[0][0]
+                    rc.execute("insert into doc.rx values (1)")
+                    time.sleep(1) # replication delay...
+                    c.execute("select count(*) from doc.rx")
+                    self.assertEqual(c.fetchall()[0][0], count + 1)
+
         # Finally validate that all shards (primaries and replicas) of all partitions are started
         # and writes into the partitioned table while upgrading were successful
         with connect(cluster.node().http_url, error_trace=True) as conn: