diff --git a/fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java b/fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java
index 7a392f4afa..a6a73354f9 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/write/DynamicPartitionCreator.java
@@ -23,6 +23,7 @@
 import org.apache.fluss.exception.PartitionNotExistException;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.utils.AutoPartitionStrategy;
 import org.apache.fluss.utils.ExceptionUtils;
@@ -66,9 +67,7 @@ public DynamicPartitionCreator(
     }
 
     public void checkAndCreatePartitionAsync(
-            PhysicalTablePath physicalTablePath,
-            List<String> partitionKeys,
-            AutoPartitionStrategy autoPartitionStrategy) {
+            PhysicalTablePath physicalTablePath, TableInfo tableInfo) {
         String partitionName = physicalTablePath.getPartitionName();
         if (partitionName == null) {
             // no need to check and create partition
@@ -87,7 +86,11 @@ public void checkAndCreatePartitionAsync(
                 // if the partition exists, we should skip creating it.
                 LOG.debug("Partition {} already exists, skipping.", physicalTablePath);
             } else {
-                // Validate early, before touching any state.
+                // Validate early, before touching any state. The strategy is only resolved here,
+                // on the partition-creation path, not on the common "already exists" path.
+                List<String> partitionKeys = tableInfo.getPartitionKeys();
+                AutoPartitionStrategy autoPartitionStrategy =
+                        tableInfo.getTableConfig().getAutoPartitionStrategy();
                 ResolvedPartitionSpec resolvedPartitionSpec =
                         ResolvedPartitionSpec.fromPartitionName(partitionKeys, partitionName);
                 validateAutoPartitionTime(
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java b/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
index b4c96b0ac4..ff8b9232f9 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java
@@ -176,10 +176,10 @@ private void doSend(WriteRecord record, WriteCallback callback) {
 
             TableInfo tableInfo = record.getTableInfo();
             PhysicalTablePath physicalTablePath = record.getPhysicalTablePath();
-            dynamicPartitionCreator.checkAndCreatePartitionAsync(
-                    physicalTablePath,
-                    tableInfo.getPartitionKeys(),
-                    tableInfo.getTableConfig().getAutoPartitionStrategy());
+            // Skip the call entirely on non-partitioned tables; there is no partition to create.
+            if (tableInfo.isPartitioned()) {
+                dynamicPartitionCreator.checkAndCreatePartitionAsync(physicalTablePath, tableInfo);
+            }
 
             // maybe create bucket assigner.
             Cluster cluster = metadataUpdater.getCluster();