alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-1684 Make sure publication initialized prior to slot

Browse Source 
This is to avoid the need to restart the connector when streaming begins due to
an error that the publication can't be found.  It would seem based on testing
that since we create the replication slot early on when it does not exist, then
when streaming starts its confused about trying to marry a newly created
publication to an slot.  By creating the publication before the slot, this
works around the problem.
This commit is contained in:
Chris Cranford 2020-01-08 11:28:16 -05:00 committed by Gunnar Morling
parent eacc624001
commit cf60466b02
1 changed files with 7 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -253,10 +253,12 @@ public ReplicationStream startStreaming(Long offset) throws SQLException, Interr
@Override @Override
public void initConnection() throws SQLException, InterruptedException { public void initConnection() throws SQLException, InterruptedException {
// See https://www.postgresql.org/docs/current/logical-replication-quick-setup.html
// For pgoutput specifically, the publication must be created before the slot.
initPublication();
if (!hasInitedSlot) { if (!hasInitedSlot) {
initReplicationSlot(); initReplicationSlot();
} }
initPublication();
} }
@Override @Override
@ -278,6 +280,10 @@ public Optional<SlotCreationResult> createReplicationSlot() throws SQLException
tempPart = "TEMPORARY"; tempPart = "TEMPORARY";
} }
// See https://www.postgresql.org/docs/current/logical-replication-quick-setup.html
// For pgoutput specifically, the publication must be created prior to the slot.
initPublication();
try (Statement stmt = pgConnection().createStatement()) { try (Statement stmt = pgConnection().createStatement()) {
String createCommand = String.format( String createCommand = String.format(
"CREATE_REPLICATION_SLOT %s %s LOGICAL %s", "CREATE_REPLICATION_SLOT %s %s LOGICAL %s",