diff --git a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/ReadonlyConfig.java b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/ReadonlyConfig.java
index 8cd058f0248..64ff5a05d38 100644
--- a/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/ReadonlyConfig.java
+++ b/seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/ReadonlyConfig.java
@@ -106,7 +106,7 @@ public <T> Optional<T> getOptional(Option<T> option) {
             }
         }
         if (value == null) {
-            return Optional.ofNullable(option.defaultValue());
+            return Optional.empty();
         }
         return Optional.of(convertValue(value, option));
     }
diff --git a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/ReadableConfigTest.java b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/ReadableConfigTest.java
index 88efee07720..b1436edd0fc 100644
--- a/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/ReadableConfigTest.java
+++ b/seatunnel-api/src/test/java/org/apache/seatunnel/api/configuration/ReadableConfigTest.java
@@ -34,7 +34,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
 @SuppressWarnings("checkstyle:StaticVariableName")
 public class ReadableConfigTest {
@@ -299,18 +298,4 @@ public void testFallbackKey() {
         readonlyConfig = ReadonlyConfig.fromMap(map);
         Assertions.assertEquals("ark", readonlyConfig.get(usernameOption));
     }
-
-    @Test
-    public void testOptional() {
-        Optional<String> notDefaultVal =
-                ReadonlyConfig.fromMap(new HashMap<>())
-                        .getOptional(Options.key("option.not-exist").stringType().noDefaultValue());
-        Assertions.assertFalse(notDefaultVal.isPresent());
-        Optional<String> defaultVal =
-                ReadonlyConfig.fromMap(new HashMap<>())
-                        .getOptional(
-                                Options.key("option.exist").stringType().defaultValue("default"));
-        Assertions.assertTrue(defaultVal.isPresent());
-        Assertions.assertEquals("default", defaultVal.get());
-    }
 }
diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSinkConfig.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSinkConfig.java
index af24a9a6b03..874eb807f33 100644
--- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSinkConfig.java
+++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcSinkConfig.java
@@ -51,11 +51,10 @@ public static JdbcSinkConfig of(ReadonlyConfig config) {
         config.getOptional(JdbcOptions.PRIMARY_KEYS).ifPresent(builder::primaryKeys);
         config.getOptional(JdbcOptions.DATABASE).ifPresent(builder::database);
         config.getOptional(JdbcOptions.TABLE).ifPresent(builder::table);
-        config.getOptional(ENABLE_UPSERT).ifPresent(builder::enableUpsert);
-        config.getOptional(IS_PRIMARY_KEY_UPDATED).ifPresent(builder::isPrimaryKeyUpdated);
-        config.getOptional(SUPPORT_UPSERT_BY_INSERT_ONLY)
-                .ifPresent(builder::supportUpsertByInsertOnly);
-        config.getOptional(JdbcOptions.QUERY).ifPresent(builder::simpleSql);
+        builder.enableUpsert(config.get(ENABLE_UPSERT));
+        builder.isPrimaryKeyUpdated(config.get(IS_PRIMARY_KEY_UPDATED));
+        builder.supportUpsertByInsertOnly(config.get(SUPPORT_UPSERT_BY_INSERT_ONLY));
+        builder.simpleSql(config.get(JdbcOptions.QUERY));
         return builder.build();
     }
 }