[DBZ] Disable IMPLICIT checkpointing support in connector

src/main/java/io/debezium/connector/yugabytedb/YugabyteDBConnectorConfig.java

@@ -731,6 +731,12 @@ public JdbcConfiguration getJdbcConfig() {
             .withImportance(Importance.LOW)
             .withDefault(false);
 
+    public static final Field FORCE_USE_IMPLICIT_STREAM = Field.create("force.use.implicit.stream")
+              .withDisplayName("Flag to forcefully use implicit streams in connector")
+              .withType(Type.BOOLEAN)
+              .withImportance(Importance.LOW)
+              .withDefault(false);
+
     public static final Field CHAR_SET = Field.create(TASK_CONFIG_PREFIX + "charset")
             .withDisplayName("YugabyteDB charset")
             .withType(ConfigDef.Type.STRING)
@@ -1268,6 +1274,10 @@ public boolean ignoreExceptions() {
         return getConfig().getBoolean(IGNORE_EXCEPTIONS);
     }
 
+    public boolean forceUseImplicitStream() {
+        return getConfig().getBoolean(FORCE_USE_IMPLICIT_STREAM);
+    }
+
     public int maxNumTablets() {
         return getConfig().getInteger(MAX_NUM_TABLETS);
     }

src/main/java/io/debezium/connector/yugabytedb/YugabyteDBgRPCConnector.java

@@ -173,6 +173,14 @@ public List<Map<String, String>> taskConfigs(int maxTasks) {
             throw new DebeziumException(e);
         }
 
+        if (!enableExplicitCheckpointing && !this.yugabyteDBConnectorConfig.forceUseImplicitStream()) {
+            final String errorMessage = "The provided stream " + this.yugabyteDBConnectorConfig.streamId()
+                    + " is an IMPLICIT stream, create a stream with EXPLICIT checkpointing and try again."
+                    + " To forcefully use an IMPLICIT stream, set configuration property force.use.implicit.stream"
+                    + " to true";
+            throw new DebeziumException(errorMessage);
+        }
+
         if (this.yugabyteDBConnectorConfig.transactionOrdering() && !enableExplicitCheckpointing) {
             final String errorMessage = "Explicit checkpointing not enabled in consistent streaming mode, "
                     + "create a stream with explicit checkpointing and try again";

src/test/java/io/debezium/connector/yugabytedb/YugabyteDBConfigTest.java

@@ -311,6 +311,41 @@ public void throwExceptionIfExplicitCheckpointingNotConfiguredWithConsistency(bo
         assertConnectorNotRunning();
     }
 
+    @Test
+    public void throwExceptionIfImplicitStreamSpecified() throws Exception {
+        TestHelper.dropAllSchemas();
+
+        // Create a stream ID with IMPLICIT checkpointing and then deploy it in a consistent streaming setup.
+        TestHelper.execute("CREATE TABLE dummy_table (id INT PRIMARY KEY);");
+        final String dbStreamId = TestHelper.getNewDbStreamId("yugabyte", "dummy_table", false,
+          false, false, false);
+        Configuration.Builder configBuilder = TestHelper.getConfigBuilder("public.dummy_table", dbStreamId);
+
+        start(YugabyteDBgRPCConnector.class, configBuilder.build(), (success, message, error) -> {
+            assertFalse(success);
+
+            assertTrue(error.getMessage().contains("is an IMPLICIT stream, create a stream with EXPLICIT checkpointing and try again"));
+        });
+
+        assertConnectorNotRunning();
+    }
+
+    @Test
+    public void shouldWorkIfForceUseImplicitStreamIsSet() throws Exception {
+        TestHelper.dropAllSchemas();
+
+        // Create a stream ID with IMPLICIT checkpointing and then deploy it in a consistent streaming setup.
+        TestHelper.execute("CREATE TABLE dummy_table (id INT PRIMARY KEY);");
+        final String dbStreamId = TestHelper.getNewDbStreamId("yugabyte", "dummy_table", false,
+          false, false, false);
+        Configuration.Builder configBuilder = TestHelper.getConfigBuilder("public.dummy_table", dbStreamId);
+        configBuilder.with(YugabyteDBConnectorConfig.FORCE_USE_IMPLICIT_STREAM, true);
+
+        start(YugabyteDBgRPCConnector.class, configBuilder.build(), (success, message, error) -> {
+            assertTrue(success);
+        });
+    }
+
     @ParameterizedTest
     @MethodSource("io.debezium.connector.yugabytedb.TestHelper#streamTypeProviderForStreaming")
     public void throwExceptionWithIncorrectTaskCountWithTransactionOrdering(boolean consistentSnapshot, boolean useSnapshot) throws Exception {