diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java
index 8db3b643cf8b..a1a2a66bc106 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java
@@ -17,6 +17,8 @@
  */
 package org.apache.beam.sdk.io.iceberg;
 
+import static java.time.format.DateTimeFormatter.ISO_LOCAL_DATE;
+import static java.time.format.DateTimeFormatter.ISO_LOCAL_TIME;
 import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
 import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
@@ -27,6 +29,10 @@
 import java.time.LocalTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.time.chrono.IsoChronology;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.format.ResolverStyle;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.List;
@@ -88,6 +94,26 @@ private IcebergUtils() {}
           .put(EnumerationType.IDENTIFIER, Types.StringType.get())
           .build();
 
+  private static final DateTimeFormatter TIMESTAMP_SPACE_FORMATTER =
+      new DateTimeFormatterBuilder()
+          .parseCaseInsensitive()
+          .append(ISO_LOCAL_DATE)
+          .appendLiteral(' ')
+          .append(ISO_LOCAL_TIME)
+          .toFormatter()
+          .withResolverStyle(ResolverStyle.STRICT)
+          .withChronology(IsoChronology.INSTANCE);
+
+  private static final DateTimeFormatter TIMESTAMPTZ_SPACE_FORMATTER =
+      new DateTimeFormatterBuilder()
+          .append(TIMESTAMP_SPACE_FORMATTER)
+          .parseLenient()
+          .appendOffsetId()
+          .parseStrict()
+          .toFormatter()
+          .withResolverStyle(ResolverStyle.STRICT)
+          .withChronology(IsoChronology.INSTANCE);
+
   private static Schema.FieldType icebergTypeToBeamFieldType(
       final Type type, @Nullable String updateCompatibilityVersion) {
     switch (type.typeId()) {
@@ -491,7 +517,12 @@ private static Object getIcebergTimestampValue(Object beamValue, boolean shouldA
       } else if (beamValue instanceof Long) { // FieldType.INT64
         return DateTimeUtil.timestamptzFromMicros((Long) beamValue);
       } else if (beamValue instanceof String) { // FieldType.STRING
-        return OffsetDateTime.parse((String) beamValue).withOffsetSameInstant(ZoneOffset.UTC);
+        String value = (String) beamValue;
+        OffsetDateTime offsetDateTime =
+            hasSpaceSeparator(value)
+                ? OffsetDateTime.parse(value, TIMESTAMPTZ_SPACE_FORMATTER)
+                : OffsetDateTime.parse(value);
+        return offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC);
       } else {
         throw new UnsupportedOperationException(
             "Unsupported Beam type for Iceberg timestamp with timezone: " + beamValue.getClass());
@@ -510,13 +541,20 @@ private static Object getIcebergTimestampValue(Object beamValue, boolean shouldA
     } else if (beamValue instanceof Long) { // FieldType.INT64
       return DateTimeUtil.timestampFromMicros((Long) beamValue);
     } else if (beamValue instanceof String) { // FieldType.STRING
-      return LocalDateTime.parse((String) beamValue);
+      String value = (String) beamValue;
+      return hasSpaceSeparator(value)
+          ? LocalDateTime.parse(value, TIMESTAMP_SPACE_FORMATTER)
+          : LocalDateTime.parse(value);
     } else {
       throw new UnsupportedOperationException(
           "Unsupported Beam type for Iceberg timestamp with timezone: " + beamValue.getClass());
     }
   }
 
+  private static boolean hasSpaceSeparator(String timestamp) {
+    return timestamp.length() > 10 && timestamp.charAt(10) == ' ';
+  }
+
   /** Converts a {@link StructLike} to a Beam {@link Row}. */
   public static Row structToRow(Schema schema, StructLike struct) {
     checkState(
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java
index aa8f6799a3eb..3ebe22e93c66 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergUtilsTest.java
@@ -35,6 +35,7 @@
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.time.format.DateTimeParseException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -149,6 +150,13 @@ private void checkRowValueToRecordValue(
       assertThat(record.getField("v"), equalTo(destValue));
     }
 
+    private Record toIcebergRecord(Type destType, String sourceValue) {
+      Schema beamSchema = Schema.of(Schema.Field.of("v", Schema.FieldType.STRING));
+      Row row = Row.withSchema(beamSchema).addValues(sourceValue).build();
+      return IcebergUtils.beamRowToIcebergRecord(
+          new org.apache.iceberg.Schema(required(0, "v", destType)), row);
+    }
+
     @Test
     public void testBoolean() {
       checkRowValueToRecordValue(Schema.FieldType.BOOLEAN, Types.BooleanType.get(), true);
@@ -225,6 +233,13 @@ public void testTimestamp() {
       LocalDateTime localDateTime = LocalDateTime.of(2024, 10, 8, 13, 18, 20, 53_000_000);
       checkRowValueToRecordValue(
           Schema.FieldType.STRING, val, Types.TimestampType.withoutZone(), localDateTime);
+
+      // Schema.FieldType.STRING, with a space separating date and time
+      checkRowValueToRecordValue(
+          Schema.FieldType.STRING,
+          "2024-10-08 13:18:20.053",
+          Types.TimestampType.withoutZone(),
+          localDateTime);
     }
 
     @Test
@@ -268,6 +283,49 @@ public void testTimestampWithZone() {
           val,
           Types.TimestampType.withZone(),
           offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC));
+
+      // Schema.FieldType.STRING, with a space separating date and time
+      checkRowValueToRecordValue(
+          Schema.FieldType.STRING,
+          "2024-10-08 13:18:20.053+03:27",
+          Types.TimestampType.withZone(),
+          offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC));
+      checkRowValueToRecordValue(
+          Schema.FieldType.STRING,
+          "2024-10-08 09:51:20.053Z",
+          Types.TimestampType.withZone(),
+          offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC));
+    }
+
+    @Test
+    public void testTimestampStringWithInvalidSeparator() {
+      for (String val :
+          Arrays.asList(
+              "2024-10-08 T13:18:20.053", "2024-10-0813:18:20.053", "2024-10-08  13:18:20.053")) {
+        assertThrows(
+            DateTimeParseException.class,
+            () -> toIcebergRecord(Types.TimestampType.withoutZone(), val));
+        assertThrows(
+            DateTimeParseException.class,
+            () -> toIcebergRecord(Types.TimestampType.withZone(), val + "Z"));
+      }
+    }
+
+    @Test
+    public void testTimestampStringWithInvalidDate() {
+      for (String val :
+          Arrays.asList(
+              "2023-02-29T13:18:20",
+              "2023-02-29 13:18:20",
+              "2024-04-31T13:18:20",
+              "2024-04-31 13:18:20")) {
+        assertThrows(
+            DateTimeParseException.class,
+            () -> toIcebergRecord(Types.TimestampType.withoutZone(), val));
+        assertThrows(
+            DateTimeParseException.class,
+            () -> toIcebergRecord(Types.TimestampType.withZone(), val + "Z"));
+      }
     }
 
     @Test