[DBZ-PGYB] Bug fix for inability to alter publication upon transitioning to streaming (#174)

## Problem

When #172 got merged, we added the logic to begin a read only
transaction so that we can use `USE_SNAPSHOT` and capture the snapshot
from the same. However, upon transition to streaming, when
`PostgresReplicationConnection#startStreaming` during streaming phase,
it internally calls `PostgresReplicationConnection#initConnection` and
then ``PostgresReplicationConnection#initPublication` which tries to
execute an `ALTER` command on the publication.

The `ALTER` command fails owing to the fact that we're still in a
read-only transaction opened at the time of slot creation. The exception
is similar to:

```
Caused by: org.apache.kafka.connect.errors.ConnectException: Unable to update filtered publication dbz_publication for "public"."test"
	at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.createOrUpdatePublicationModeFilterted(PostgresReplicationConnection.java:232)
	at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.initPublication(PostgresReplicationConnection.java:196)
	at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.initConnection(PostgresReplicationConnection.java:496)
	at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.startStreaming(PostgresReplicationConnection.java:396)
	at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:182)
	... 9 more
Caused by: com.yugabyte.util.PSQLException: ERROR: cannot execute ALTER PUBLICATION in a read-only transaction
	at com.yugabyte.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2675)
	at com.yugabyte.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2365)
	at com.yugabyte.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:355)
	at com.yugabyte.jdbc.PgStatement.executeInternal(PgStatement.java:490)
	at com.yugabyte.jdbc.PgStatement.execute(PgStatement.java:408)
	at com.yugabyte.jdbc.PgStatement.executeWithFlags(PgStatement.java:329)
	at com.yugabyte.jdbc.PgStatement.executeCachedSql(PgStatement.java:315)
	at com.yugabyte.jdbc.PgStatement.executeWithFlags(PgStatement.java:291)
	at com.yugabyte.jdbc.PgStatement.execute(PgStatement.java:286)
	at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.createOrUpdatePublicationModeFilterted(PostgresReplicationConnection.java:229)
```

## Solution

As a fix, we will now commit at the time of snapshot completion and
rollback at the time of snapshot abortion so that we ensure that we are
not going into the streaming phase with any open transaction. Another
added guardrail in this PR is that we only open the read-only
transaction when `streaming.mode` is set to `parallel`.

Author: vaibhav-yb

debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java

@@ -14,6 +14,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import io.debezium.DebeziumException;
 import io.debezium.pipeline.spi.ChangeRecordEmitter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -260,11 +261,25 @@ protected SchemaChangeEvent getCreateTableEvent(RelationalSnapshotContext<Postgr
 
     @Override
     protected void completed(SnapshotContext<PostgresPartition, PostgresOffsetContext> snapshotContext) {
+        try {
+            jdbcConnection.commit();
+        } catch (SQLException sqle) {
+            LOGGER.error("Exception while committing prior to reporting snapshot completion {}", sqle);
+            throw new DebeziumException(sqle);
+        }
+
         snapshotter.snapshotCompleted();
     }
 
     @Override
     protected void aborted(SnapshotContext<PostgresPartition, PostgresOffsetContext> snapshotContext) {
+        try {
+            jdbcConnection.rollback();
+        } catch (SQLException sqle) {
+            LOGGER.error("Exception while rolling back prior to reporting snapshot abortion {}", sqle);
+            throw new DebeziumException(sqle);
+        }
+
         snapshotter.snapshotAborted();
     }
 

debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java

@@ -533,8 +533,14 @@ public Optional<SlotCreationResult> createReplicationSlot() throws SQLException
                     plugin.getPostgresPluginName(),
                     lsnType.getLsnTypeName(),
                     streamingMode.isParallel() ? "USE_SNAPSHOT" : "");
-            LOGGER.info("executing: BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY");
-            stmt.execute("BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY");
+
+            // Begin a read-only transaction when it is the parallel streaming mode because
+            // we will be using this read-only transaction to take the snapshot further.
+            if (connectorConfig.streamingMode().isParallel() ) {
+                LOGGER.info("executing: BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY");
+                stmt.execute("BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY");
+            }
+
             LOGGER.info("Creating replication slot with command {}", createCommand);
             stmt.execute(createCommand);
             // when we are in Postgres 9.4+, we can parse the slot creation info,