Repository: pgjdbc/pgjdbc
Pull request #1306 (2018-10-03 22:22:50)

feat: Extend ReplicationCreateSlotBuilder DSL to support temporary replications slots
With Postgres 10 it is possible to create temporary replication slots.
The Postgres JDBC Driver provides a DSL to create replication slots, but there was no option to declare the slot as temporary.
This PR extends the DSL by a withTemporaryOption method.

BREAKING CHANGE: AbstractCreateSlotBuilder has no parameterless constructor anymore
Closes https://github.com/pgjdbc/pgjdbc/issues/1305

### All Submissions:

* [x] Have you followed the guidelines in our [Contributing](CONTRIBUTING.md) document?
* [x] Have you checked to ensure there aren't other open [Pull Requests](../../pulls) for the same update/change?

<!-- You can erase any parts of this template not applicable to your Pull Request. -->

### New Feature Submissions:

1. [x] Does your submission pass tests?
2. [x] Does mvn checkstyle:check pass ?

### Changes to Existing Features:

* [x] Does this break existing behaviour? If so please explain.
The `AbstractCreateSlotBuilder` constructor now has a `BaseConnection` parameter which is required to check the server version in the new `withTemporaryOption` method.

* [x] Have you added an explanation of what your changes do and why you'd like us to include them?
* [x] Have you written new tests for your core changes, as applicable?
* [x] Have you successfully run tests with your changes locally?


Patch:
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/fluent/AbstractCreateSlotBuilder.java b/pgjdbc/src/main/java/org/postgresql/replication/fluent/AbstractCreateSlotBuilder.java
index 54f946b4..c55f65b8 100644
--- a/pgjdbc/src/main/java/org/postgresql/replication/fluent/AbstractCreateSlotBuilder.java
+++ b/pgjdbc/src/main/java/org/postgresql/replication/fluent/AbstractCreateSlotBuilder.java
@@ -5,10 +5,22 @@
 
 package org.postgresql.replication.fluent;
 
+import org.postgresql.core.BaseConnection;
+import org.postgresql.core.ServerVersion;
+import org.postgresql.util.GT;
+
+import java.sql.SQLFeatureNotSupportedException;
+
 public abstract class AbstractCreateSlotBuilder<T extends ChainedCommonCreateSlotBuilder<T>>
     implements ChainedCommonCreateSlotBuilder<T> {
 
   protected String slotName;
+  protected boolean temporaryOption = false;
+  protected BaseConnection connection;
+
+  protected AbstractCreateSlotBuilder(BaseConnection connection) {
+    this.connection = connection;
+  }
 
   protected abstract T self();
 
@@ -17,4 +29,17 @@ public abstract class AbstractCreateSlotBuilder<T extends ChainedCommonCreateSlo
     this.slotName = slotName;
     return self();
   }
+
+  @Override
+  public T withTemporaryOption() throws SQLFeatureNotSupportedException {
+
+    if (!connection.haveMinimumServerVersion(ServerVersion.v10)) {
+      throw new SQLFeatureNotSupportedException(
+          GT.tr("Server does not support temporary replication slots")
+      );
+    }
+
+    this.temporaryOption = true;
+    return self();
+  }
 }
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/fluent/ChainedCommonCreateSlotBuilder.java b/pgjdbc/src/main/java/org/postgresql/replication/fluent/ChainedCommonCreateSlotBuilder.java
index 78a4565f..4076a4b7 100644
--- a/pgjdbc/src/main/java/org/postgresql/replication/fluent/ChainedCommonCreateSlotBuilder.java
+++ b/pgjdbc/src/main/java/org/postgresql/replication/fluent/ChainedCommonCreateSlotBuilder.java
@@ -6,6 +6,7 @@
 package org.postgresql.replication.fluent;
 
 import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
 
 /**
  * Fluent interface for specify common parameters for create Logical and Physical replication slot.
@@ -22,6 +23,17 @@ public interface ChainedCommonCreateSlotBuilder<T extends ChainedCommonCreateSlo
    */
   T withSlotName(String slotName);
 
+  /**
+   * <p>Temporary slots are not saved to disk and are automatically dropped on error or when
+   * the session has finished.</p>
+   *
+   * <p>This feature is only supported by PostgreSQL versions &gt;= 10.</p>
+   *
+   * @return T a slot builder
+   * @throws SQLFeatureNotSupportedException thrown if PostgreSQL version is less than 10.
+   */
+  T withTemporaryOption() throws SQLFeatureNotSupportedException;
+
   /**
    * Create slot with specified parameters in database.
    * @throws SQLException on error
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/fluent/logical/LogicalCreateSlotBuilder.java b/pgjdbc/src/main/java/org/postgresql/replication/fluent/logical/LogicalCreateSlotBuilder.java
index 0369976c..72443c18 100644
--- a/pgjdbc/src/main/java/org/postgresql/replication/fluent/logical/LogicalCreateSlotBuilder.java
+++ b/pgjdbc/src/main/java/org/postgresql/replication/fluent/logical/LogicalCreateSlotBuilder.java
@@ -15,10 +15,9 @@ public class LogicalCreateSlotBuilder
     extends AbstractCreateSlotBuilder<ChainedLogicalCreateSlotBuilder>
     implements ChainedLogicalCreateSlotBuilder {
   private String outputPlugin;
-  private BaseConnection connection;
 
   public LogicalCreateSlotBuilder(BaseConnection connection) {
-    this.connection = connection;
+    super(connection);
   }
 
   @Override
@@ -45,7 +44,12 @@ public class LogicalCreateSlotBuilder
 
     Statement statement = connection.createStatement();
     try {
-      statement.execute(String.format("CREATE_REPLICATION_SLOT %s LOGICAL %s", slotName, outputPlugin));
+      statement.execute(String.format(
+              "CREATE_REPLICATION_SLOT %s %s LOGICAL %s",
+              slotName,
+              temporaryOption ? "TEMPORARY" : "",
+              outputPlugin
+      ));
     } finally {
       statement.close();
     }
diff --git a/pgjdbc/src/main/java/org/postgresql/replication/fluent/physical/PhysicalCreateSlotBuilder.java b/pgjdbc/src/main/java/org/postgresql/replication/fluent/physical/PhysicalCreateSlotBuilder.java
index 9d99fa4e..1ae0ff0f 100644
--- a/pgjdbc/src/main/java/org/postgresql/replication/fluent/physical/PhysicalCreateSlotBuilder.java
+++ b/pgjdbc/src/main/java/org/postgresql/replication/fluent/physical/PhysicalCreateSlotBuilder.java
@@ -14,10 +14,9 @@ import java.sql.Statement;
 public class PhysicalCreateSlotBuilder
     extends AbstractCreateSlotBuilder<ChainedPhysicalCreateSlotBuilder>
     implements ChainedPhysicalCreateSlotBuilder {
-  private BaseConnection connection;
 
   public PhysicalCreateSlotBuilder(BaseConnection connection) {
-    this.connection = connection;
+    super(connection);
   }
 
   @Override
@@ -33,7 +32,11 @@ public class PhysicalCreateSlotBuilder
 
     Statement statement = connection.createStatement();
     try {
-      statement.execute(String.format("CREATE_REPLICATION_SLOT %s PHYSICAL", slotName));
+      statement.execute(String.format(
+              "CREATE_REPLICATION_SLOT %s %s PHYSICAL",
+              slotName,
+              temporaryOption ? "TEMPORARY" : ""
+      ));
     } finally {
       statement.close();
     }


Test patch:
diff --git a/pgjdbc/src/test/java/org/postgresql/replication/ReplicationSlotTest.java b/pgjdbc/src/test/java/org/postgresql/replication/ReplicationSlotTest.java
index 2e9b1d73..961075d2 100644
--- a/pgjdbc/src/test/java/org/postgresql/replication/ReplicationSlotTest.java
+++ b/pgjdbc/src/test/java/org/postgresql/replication/ReplicationSlotTest.java
@@ -7,9 +7,13 @@ package org.postgresql.replication;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.fail;
+import static org.junit.Assume.assumeFalse;
+import static org.junit.Assume.assumeTrue;
 
 import org.postgresql.PGConnection;
 import org.postgresql.PGProperty;
+import org.postgresql.core.BaseConnection;
+import org.postgresql.core.ServerVersion;
 import org.postgresql.test.TestUtil;
 import org.postgresql.test.util.rules.ServerVersionRule;
 import org.postgresql.test.util.rules.annotation.HaveMinimalServerVersion;
@@ -23,6 +27,7 @@ import org.junit.Test;
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
 import java.sql.Statement;
 import java.util.Properties;
 
@@ -80,7 +85,67 @@ public class ReplicationSlotTest {
 
     boolean result = isPhysicalSlotExists(slotName);
 
-    assertThat(result, CoreMatchers.equalTo(true));
+    assertThat("Slot should exist", result, CoreMatchers.equalTo(true));
+
+    result = isSlotTemporary(slotName);
+
+    assertThat("Slot should not be temporary by default", result, CoreMatchers.equalTo(false));
+  }
+
+  @Test
+  public void testCreateTemporaryPhysicalSlotPg10AndHigher()
+      throws SQLException {
+    assumeTrue(TestUtil.haveMinimumServerVersion(replConnection, ServerVersion.v10));
+
+    BaseConnection baseConnection = (BaseConnection) replConnection;
+
+    String slotName = "pgjdbc_test_create_temporary_physical_replication_slot_pg_10_or_higher";
+
+    try {
+
+      baseConnection
+          .getReplicationAPI()
+          .createReplicationSlot()
+          .physical()
+          .withSlotName(slotName)
+          .withTemporaryOption()
+          .make();
+
+    } catch (SQLFeatureNotSupportedException e) {
+
+      fail("PostgreSQL >= 10 should support temporary replication slots");
+
+    }
+
+    boolean result = isSlotTemporary(slotName);
+
+    assertThat("Slot is not temporary", result, CoreMatchers.equalTo(true));
+  }
+
+  @Test
+  public void testCreateTemporaryPhysicalSlotPgLowerThan10()
+      throws SQLException {
+    assumeFalse(TestUtil.haveMinimumServerVersion(replConnection, ServerVersion.v10));
+
+    BaseConnection baseConnection = (BaseConnection) replConnection;
+
+    String slotName = "pgjdbc_test_create_temporary_physical_replication_slot_pg_lower_than_10";
+
+    try {
+
+      baseConnection
+          .getReplicationAPI()
+          .createReplicationSlot()
+          .physical()
+          .withSlotName(slotName)
+          .withTemporaryOption()
+          .make();
+
+      fail("PostgreSQL < 10 does not support temporary replication slots");
+
+    } catch (SQLFeatureNotSupportedException e) {
+      // success
+    }
   }
 
   @Test
@@ -151,7 +216,69 @@ public class ReplicationSlotTest {
 
     boolean result = isLogicalSlotExists(slotName);
 
-    assertThat(result, CoreMatchers.equalTo(true));
+    assertThat("Slot should exist", result, CoreMatchers.equalTo(true));
+
+    result = isSlotTemporary(slotName);
+
+    assertThat("Slot should not be temporary by default", result, CoreMatchers.equalTo(false));
+  }
+
+  @Test
+  public void testCreateTemporaryLogicalSlotPg10AndHigher()
+      throws SQLException {
+    assumeTrue(TestUtil.haveMinimumServerVersion(replConnection, ServerVersion.v10));
+
+    BaseConnection baseConnection = (BaseConnection) replConnection;
+
+    String slotName = "pgjdbc_test_create_temporary_logical_replication_slot_pg_10_or_higher";
+
+    try {
+
+      baseConnection
+          .getReplicationAPI()
+          .createReplicationSlot()
+          .logical()
+          .withSlotName(slotName)
+          .withOutputPlugin("test_decoding")
+          .withTemporaryOption()
+          .make();
+
+    } catch (SQLFeatureNotSupportedException e) {
+
+      fail("PostgreSQL >= 10 should support temporary replication slots");
+
+    }
+
+    boolean result = isSlotTemporary(slotName);
+
+    assertThat("Slot is not temporary", result, CoreMatchers.equalTo(true));
+  }
+
+  @Test
+  public void testCreateTemporaryLogicalSlotPgLowerThan10()
+      throws SQLException {
+    assumeFalse(TestUtil.haveMinimumServerVersion(replConnection, ServerVersion.v10));
+
+    BaseConnection baseConnection = (BaseConnection) replConnection;
+
+    String slotName = "pgjdbc_test_create_temporary_logical_replication_slot_pg_lower_than_10";
+
+    try {
+
+      baseConnection
+          .getReplicationAPI()
+          .createReplicationSlot()
+          .logical()
+          .withSlotName(slotName)
+          .withOutputPlugin("test_decoding")
+          .withTemporaryOption()
+          .make();
+
+      fail("PostgreSQL < 10 does not support temporary replication slots");
+
+    } catch (SQLFeatureNotSupportedException e) {
+      // success
+    }
   }
 
   @Test
@@ -205,6 +332,23 @@ public class ReplicationSlotTest {
     return result;
   }
 
+  private boolean isSlotTemporary(String slotName) throws SQLException {
+    if (!TestUtil.haveMinimumServerVersion(sqlConnection, ServerVersion.v10)) {
+      return false;
+    }
+
+    boolean result;
+
+    Statement st = sqlConnection.createStatement();
+    ResultSet resultSet = st.executeQuery(
+            "select 1 from pg_replication_slots where slot_name = '" + slotName
+                    + "' and temporary = true");
+    result = resultSet.next();
+    resultSet.close();
+    st.close();
+    return result;
+  }
+
   private void dropReplicationSlot() throws Exception {
     if (slotName != null) {
       TestUtil.dropReplicationSlot(sqlConnection, slotName);