diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
index c0c467fc8..cb692ce31 100644
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.connector.jdbc.catalog;
 
+import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory;
 import org.apache.flink.table.api.Schema;
 import org.apache.flink.table.api.ValidationException;
@@ -87,6 +88,7 @@
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /** Abstract catalog for any JDBC catalogs. */
+@PublicEvolving
 public abstract class AbstractJdbcCatalog extends AbstractCatalog {
 
     private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcCatalog.class);
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/AbstractDialect.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/AbstractDialect.java
index ae710a331..d838ebe6d 100644
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/AbstractDialect.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/AbstractDialect.java
@@ -19,7 +19,6 @@
 package org.apache.flink.connector.jdbc.dialect;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.jdbc.catalog.AbstractJdbcCatalog;
 import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.LogicalTypeRoot;
@@ -30,7 +29,6 @@
 
 import java.util.Arrays;
 import java.util.Optional;
-import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -47,17 +45,6 @@
 @PublicEvolving
 public abstract class AbstractDialect implements JdbcDialect {
 
-    @Override
-    public AbstractJdbcCatalog createCatalog(
-            ClassLoader userClassLoader,
-            String catalogName,
-            String defaultDatabase,
-            String baseUrl,
-            Properties connectionProperties) {
-        throw new UnsupportedOperationException(
-                String.format("Catalog for '%s' is not supported yet.", dialectName()));
-    }
-
     @Override
     public void validate(RowType rowType) throws ValidationException {
         for (RowType.RowField field : rowType.getFields()) {
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialect.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialect.java
index d6e7a777b..5438a95b5 100644
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialect.java
+++ b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialect.java
@@ -54,12 +54,15 @@ public interface JdbcDialect extends Serializable {
      * @param connectionProperties the properties used to connect the database
      * @return A JDBC catalog instance.
      */
-    AbstractJdbcCatalog createCatalog(
+    default AbstractJdbcCatalog createCatalog(
             ClassLoader userClassLoader,
             String catalogName,
             String defaultDatabase,
             String baseUrl,
-            Properties connectionProperties);
+            Properties connectionProperties) {
+        throw new UnsupportedOperationException(
+                String.format("Catalog for '%s' is not supported yet.", dialectName()));
+    }
 
     /**
      * Get converter that convert jdbc object and Flink internal object each other.