diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index 25efa744a733..a2d2807485d5 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
@@ -32,6 +32,7 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -78,14 +79,36 @@ private void findOrCreateTable(
       TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) {
     Tuple2<Boolean, Exception> exists = cache.exists(identifier);
     if (Boolean.FALSE.equals(exists.f0)) {
-      if (exists.f1 instanceof NoSuchNamespaceException) {
+      if (exists.f1 != null
+          && !(exists.f1 instanceof NoSuchTableException)
+          && !(exists.f1 instanceof NoSuchNamespaceException)) {
+        throw new IllegalStateException(
+            String.format("Unexpected failure looking up table %s", identifier), exists.f1);
+      }
+
+      // Create the namespace if it is missing. Tasks that get here at the same time can all find it
+      // missing and try to create it, which is not an error. The lookup of the table is not used to
+      // tell: which exception it throws for a missing namespace is up to the catalog (Glue throws
+      // NoSuchTableException), and an answer from the cache carries none.
+      if (catalog instanceof SupportsNamespaces) {
         SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog;
-        LOG.info("Namespace {} not found during table search. Creating namespace", identifier);
-        try {
-          catalogWithNameSpace.createNamespace(identifier.namespace());
-        } catch (AlreadyExistsException e) {
-          LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
+        // A table without a namespace has none to create, and some catalogs, like the Hadoop
+        // catalog, do not accept the empty namespace in the call.
+        if (!identifier.namespace().isEmpty()
+            && !catalogWithNameSpace.namespaceExists(identifier.namespace())) {
+          LOG.info("Namespace {} not found during table search. Creating namespace", identifier);
+          try {
+            catalogWithNameSpace.createNamespace(identifier.namespace());
+          } catch (AlreadyExistsException e) {
+            LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
+          }
         }
+      } else {
+        throw new UnsupportedOperationException(
+            String.format(
+                "Cannot create table %s: catalog %s does not support namespaces, so the namespace of the table cannot be created if it is missing",
+                identifier, catalog.name()),
+            exists.f1);
       }
 
       LOG.info("Table {} not found during table search. Creating table.", identifier);
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
index 367250cf278c..b8663e6e8ad1 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
@@ -19,10 +19,13 @@
 package org.apache.iceberg.flink.sink.dynamic;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.nio.file.Path;
 import java.time.LocalDate;
+import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
@@ -35,6 +38,8 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.flink.sink.TestFlinkIcebergSinkBase;
 import org.apache.iceberg.inmemory.InMemoryCatalog;
 import org.apache.iceberg.types.Types;
@@ -158,6 +163,148 @@ void testBranchCreationAndCaching() {
     assertThat(cache.getInternalCache()).contains(Map.entry(tableIdentifier, cacheItem));
   }
 
+  @Test
+  void testNamespaceIsCreatedWhenTheCacheAlreadyHoldsTheMissingTable() {
+    // A catalog that reports a table in a missing namespace as NoSuchNamespaceException.
+    InMemoryCatalog catalog =
+        new InMemoryCatalog() {
+          @Override
+          public Table loadTable(TableIdentifier identifier) {
+            if (!namespaceExists(identifier.namespace())) {
+              throw new NoSuchNamespaceException(
+                  "Namespace does not exist: %s", identifier.namespace());
+            }
+
+            return super.loadTable(identifier);
+          }
+        };
+    catalog.initialize("catalog", Map.of());
+    Namespace namespace = Namespace.of("cached_namespace");
+    TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    // The record processor asks the cache whether the table exists before it calls the updater, so
+    // the
+    // updater is given the cached answer, which does not carry the exception of the lookup.
+    Tuple2<Boolean, Exception> first = cache.exists(tableIdentifier);
+    assertThat(first.f0).isFalse();
+    assertThat(first.f1).isInstanceOf(NoSuchNamespaceException.class);
+    assertThat(cache.exists(tableIdentifier).f1).isNull();
+
+    tableUpdater.update(
+        tableIdentifier,
+        SnapshotRef.MAIN_BRANCH,
+        SCHEMA,
+        PartitionSpec.unpartitioned(),
+        TableCreator.DEFAULT);
+
+    assertThat(catalog.namespaceExists(namespace)).isTrue();
+    assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+  }
+
+  private static Catalog catalogWithoutNamespaceSupport(Supplier<RuntimeException> lookupFailure) {
+    return new Catalog() {
+      @Override
+      public List<TableIdentifier> listTables(Namespace namespace) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public Table loadTable(TableIdentifier identifier) {
+        throw lookupFailure.get();
+      }
+
+      @Override
+      public boolean dropTable(TableIdentifier identifier, boolean purge) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public void renameTable(TableIdentifier from, TableIdentifier to) {
+        throw new UnsupportedOperationException();
+      }
+    };
+  }
+
+  @Test
+  void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() {
+    Catalog catalog =
+        catalogWithoutNamespaceSupport(
+            () -> new NoSuchNamespaceException("Namespace does not exist"));
+    TableIdentifier tableIdentifier =
+        TableIdentifier.of(Namespace.of("missing_namespace"), "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageContaining("does not support namespaces")
+        .hasCauseInstanceOf(NoSuchNamespaceException.class);
+  }
+
+  @Test
+  void testCatalogWithoutNamespaceSupportFailsWhateverTheLookupReported() {
+    Catalog catalog =
+        catalogWithoutNamespaceSupport(() -> new NoSuchTableException("Table does not exist"));
+    TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("some_namespace"), "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    // The first answer carries the exception of the lookup, the second one comes from the cache and
+    // carries none: both must fail the same way.
+    assertThat(cache.exists(tableIdentifier).f1).isInstanceOf(NoSuchTableException.class);
+    assertThat(cache.exists(tableIdentifier).f1).isNull();
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageContaining("does not support namespaces");
+  }
+
+  @Test
+  void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    TableIdentifier tableIdentifier = TableIdentifier.parse("myTable");
+    RuntimeException failure = new RuntimeException("catalog unavailable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS) {
+          @Override
+          Tuple2<Boolean, Exception> exists(TableIdentifier identifier) {
+            return Tuple2.of(false, failure);
+          }
+        };
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("Unexpected failure looking up table")
+        .hasCause(failure);
+    assertThat(catalog.tableExists(tableIdentifier)).isFalse();
+  }
+
   @Test
   void testSpecCreation() {
     Catalog catalog = CATALOG_EXTENSION.catalog();
@@ -453,4 +600,42 @@ void testTableCreationWithExistingNamespace() {
     assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
     assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME);
   }
+
+  @Test
+  void testTableCreationWhenCatalogReportsMissingNamespaceAsNoSuchTable() {
+    // The Glue catalog answers a lookup in a database that does not exist with
+    // NoSuchTableException,
+    // not NoSuchNamespaceException, and then refuses to create the table.
+    InMemoryCatalog catalog =
+        new InMemoryCatalog() {
+          @Override
+          public Table loadTable(TableIdentifier identifier) {
+            if (!namespaceExists(identifier.namespace())) {
+              throw new NoSuchTableException("Table does not exist: %s", identifier);
+            }
+
+            return super.loadTable(identifier);
+          }
+        };
+    catalog.initialize("catalog", Map.of());
+    Namespace namespace = Namespace.of("glue_like_namespace");
+    TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThat(catalog.namespaceExists(namespace)).isFalse();
+
+    Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+        tableUpdater.update(
+            tableIdentifier,
+            SnapshotRef.MAIN_BRANCH,
+            SCHEMA,
+            PartitionSpec.unpartitioned(),
+            TableCreator.DEFAULT);
+
+    assertThat(catalog.namespaceExists(namespace)).isTrue();
+    assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+    assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
+  }
 }
diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index 25efa744a733..a2d2807485d5 100644
--- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
@@ -32,6 +32,7 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -78,14 +79,36 @@ private void findOrCreateTable(
       TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) {
     Tuple2<Boolean, Exception> exists = cache.exists(identifier);
     if (Boolean.FALSE.equals(exists.f0)) {
-      if (exists.f1 instanceof NoSuchNamespaceException) {
+      if (exists.f1 != null
+          && !(exists.f1 instanceof NoSuchTableException)
+          && !(exists.f1 instanceof NoSuchNamespaceException)) {
+        throw new IllegalStateException(
+            String.format("Unexpected failure looking up table %s", identifier), exists.f1);
+      }
+
+      // Create the namespace if it is missing. Tasks that get here at the same time can all find it
+      // missing and try to create it, which is not an error. The lookup of the table is not used to
+      // tell: which exception it throws for a missing namespace is up to the catalog (Glue throws
+      // NoSuchTableException), and an answer from the cache carries none.
+      if (catalog instanceof SupportsNamespaces) {
         SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog;
-        LOG.info("Namespace {} not found during table search. Creating namespace", identifier);
-        try {
-          catalogWithNameSpace.createNamespace(identifier.namespace());
-        } catch (AlreadyExistsException e) {
-          LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
+        // A table without a namespace has none to create, and some catalogs, like the Hadoop
+        // catalog, do not accept the empty namespace in the call.
+        if (!identifier.namespace().isEmpty()
+            && !catalogWithNameSpace.namespaceExists(identifier.namespace())) {
+          LOG.info("Namespace {} not found during table search. Creating namespace", identifier);
+          try {
+            catalogWithNameSpace.createNamespace(identifier.namespace());
+          } catch (AlreadyExistsException e) {
+            LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
+          }
         }
+      } else {
+        throw new UnsupportedOperationException(
+            String.format(
+                "Cannot create table %s: catalog %s does not support namespaces, so the namespace of the table cannot be created if it is missing",
+                identifier, catalog.name()),
+            exists.f1);
       }
 
       LOG.info("Table {} not found during table search. Creating table.", identifier);
diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
index 367250cf278c..b8663e6e8ad1 100644
--- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
+++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
@@ -19,10 +19,13 @@
 package org.apache.iceberg.flink.sink.dynamic;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.nio.file.Path;
 import java.time.LocalDate;
+import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
@@ -35,6 +38,8 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.flink.sink.TestFlinkIcebergSinkBase;
 import org.apache.iceberg.inmemory.InMemoryCatalog;
 import org.apache.iceberg.types.Types;
@@ -158,6 +163,148 @@ void testBranchCreationAndCaching() {
     assertThat(cache.getInternalCache()).contains(Map.entry(tableIdentifier, cacheItem));
   }
 
+  @Test
+  void testNamespaceIsCreatedWhenTheCacheAlreadyHoldsTheMissingTable() {
+    // A catalog that reports a table in a missing namespace as NoSuchNamespaceException.
+    InMemoryCatalog catalog =
+        new InMemoryCatalog() {
+          @Override
+          public Table loadTable(TableIdentifier identifier) {
+            if (!namespaceExists(identifier.namespace())) {
+              throw new NoSuchNamespaceException(
+                  "Namespace does not exist: %s", identifier.namespace());
+            }
+
+            return super.loadTable(identifier);
+          }
+        };
+    catalog.initialize("catalog", Map.of());
+    Namespace namespace = Namespace.of("cached_namespace");
+    TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    // The record processor asks the cache whether the table exists before it calls the updater, so
+    // the
+    // updater is given the cached answer, which does not carry the exception of the lookup.
+    Tuple2<Boolean, Exception> first = cache.exists(tableIdentifier);
+    assertThat(first.f0).isFalse();
+    assertThat(first.f1).isInstanceOf(NoSuchNamespaceException.class);
+    assertThat(cache.exists(tableIdentifier).f1).isNull();
+
+    tableUpdater.update(
+        tableIdentifier,
+        SnapshotRef.MAIN_BRANCH,
+        SCHEMA,
+        PartitionSpec.unpartitioned(),
+        TableCreator.DEFAULT);
+
+    assertThat(catalog.namespaceExists(namespace)).isTrue();
+    assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+  }
+
+  private static Catalog catalogWithoutNamespaceSupport(Supplier<RuntimeException> lookupFailure) {
+    return new Catalog() {
+      @Override
+      public List<TableIdentifier> listTables(Namespace namespace) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public Table loadTable(TableIdentifier identifier) {
+        throw lookupFailure.get();
+      }
+
+      @Override
+      public boolean dropTable(TableIdentifier identifier, boolean purge) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public void renameTable(TableIdentifier from, TableIdentifier to) {
+        throw new UnsupportedOperationException();
+      }
+    };
+  }
+
+  @Test
+  void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() {
+    Catalog catalog =
+        catalogWithoutNamespaceSupport(
+            () -> new NoSuchNamespaceException("Namespace does not exist"));
+    TableIdentifier tableIdentifier =
+        TableIdentifier.of(Namespace.of("missing_namespace"), "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageContaining("does not support namespaces")
+        .hasCauseInstanceOf(NoSuchNamespaceException.class);
+  }
+
+  @Test
+  void testCatalogWithoutNamespaceSupportFailsWhateverTheLookupReported() {
+    Catalog catalog =
+        catalogWithoutNamespaceSupport(() -> new NoSuchTableException("Table does not exist"));
+    TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("some_namespace"), "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    // The first answer carries the exception of the lookup, the second one comes from the cache and
+    // carries none: both must fail the same way.
+    assertThat(cache.exists(tableIdentifier).f1).isInstanceOf(NoSuchTableException.class);
+    assertThat(cache.exists(tableIdentifier).f1).isNull();
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageContaining("does not support namespaces");
+  }
+
+  @Test
+  void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    TableIdentifier tableIdentifier = TableIdentifier.parse("myTable");
+    RuntimeException failure = new RuntimeException("catalog unavailable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS) {
+          @Override
+          Tuple2<Boolean, Exception> exists(TableIdentifier identifier) {
+            return Tuple2.of(false, failure);
+          }
+        };
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("Unexpected failure looking up table")
+        .hasCause(failure);
+    assertThat(catalog.tableExists(tableIdentifier)).isFalse();
+  }
+
   @Test
   void testSpecCreation() {
     Catalog catalog = CATALOG_EXTENSION.catalog();
@@ -453,4 +600,42 @@ void testTableCreationWithExistingNamespace() {
     assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
     assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME);
   }
+
+  @Test
+  void testTableCreationWhenCatalogReportsMissingNamespaceAsNoSuchTable() {
+    // The Glue catalog answers a lookup in a database that does not exist with
+    // NoSuchTableException,
+    // not NoSuchNamespaceException, and then refuses to create the table.
+    InMemoryCatalog catalog =
+        new InMemoryCatalog() {
+          @Override
+          public Table loadTable(TableIdentifier identifier) {
+            if (!namespaceExists(identifier.namespace())) {
+              throw new NoSuchTableException("Table does not exist: %s", identifier);
+            }
+
+            return super.loadTable(identifier);
+          }
+        };
+    catalog.initialize("catalog", Map.of());
+    Namespace namespace = Namespace.of("glue_like_namespace");
+    TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThat(catalog.namespaceExists(namespace)).isFalse();
+
+    Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+        tableUpdater.update(
+            tableIdentifier,
+            SnapshotRef.MAIN_BRANCH,
+            SCHEMA,
+            PartitionSpec.unpartitioned(),
+            TableCreator.DEFAULT);
+
+    assertThat(catalog.namespaceExists(namespace)).isTrue();
+    assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+    assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
+  }
 }
diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index 25efa744a733..a2d2807485d5 100644
--- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
@@ -32,6 +32,7 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -78,14 +79,36 @@ private void findOrCreateTable(
       TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) {
     Tuple2<Boolean, Exception> exists = cache.exists(identifier);
     if (Boolean.FALSE.equals(exists.f0)) {
-      if (exists.f1 instanceof NoSuchNamespaceException) {
+      if (exists.f1 != null
+          && !(exists.f1 instanceof NoSuchTableException)
+          && !(exists.f1 instanceof NoSuchNamespaceException)) {
+        throw new IllegalStateException(
+            String.format("Unexpected failure looking up table %s", identifier), exists.f1);
+      }
+
+      // Create the namespace if it is missing. Tasks that get here at the same time can all find it
+      // missing and try to create it, which is not an error. The lookup of the table is not used to
+      // tell: which exception it throws for a missing namespace is up to the catalog (Glue throws
+      // NoSuchTableException), and an answer from the cache carries none.
+      if (catalog instanceof SupportsNamespaces) {
         SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog;
-        LOG.info("Namespace {} not found during table search. Creating namespace", identifier);
-        try {
-          catalogWithNameSpace.createNamespace(identifier.namespace());
-        } catch (AlreadyExistsException e) {
-          LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
+        // A table without a namespace has none to create, and some catalogs, like the Hadoop
+        // catalog, do not accept the empty namespace in the call.
+        if (!identifier.namespace().isEmpty()
+            && !catalogWithNameSpace.namespaceExists(identifier.namespace())) {
+          LOG.info("Namespace {} not found during table search. Creating namespace", identifier);
+          try {
+            catalogWithNameSpace.createNamespace(identifier.namespace());
+          } catch (AlreadyExistsException e) {
+            LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
+          }
         }
+      } else {
+        throw new UnsupportedOperationException(
+            String.format(
+                "Cannot create table %s: catalog %s does not support namespaces, so the namespace of the table cannot be created if it is missing",
+                identifier, catalog.name()),
+            exists.f1);
       }
 
       LOG.info("Table {} not found during table search. Creating table.", identifier);
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
index 367250cf278c..b8663e6e8ad1 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
@@ -19,10 +19,13 @@
 package org.apache.iceberg.flink.sink.dynamic;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.nio.file.Path;
 import java.time.LocalDate;
+import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
@@ -35,6 +38,8 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.flink.sink.TestFlinkIcebergSinkBase;
 import org.apache.iceberg.inmemory.InMemoryCatalog;
 import org.apache.iceberg.types.Types;
@@ -158,6 +163,148 @@ void testBranchCreationAndCaching() {
     assertThat(cache.getInternalCache()).contains(Map.entry(tableIdentifier, cacheItem));
   }
 
+  @Test
+  void testNamespaceIsCreatedWhenTheCacheAlreadyHoldsTheMissingTable() {
+    // A catalog that reports a table in a missing namespace as NoSuchNamespaceException.
+    InMemoryCatalog catalog =
+        new InMemoryCatalog() {
+          @Override
+          public Table loadTable(TableIdentifier identifier) {
+            if (!namespaceExists(identifier.namespace())) {
+              throw new NoSuchNamespaceException(
+                  "Namespace does not exist: %s", identifier.namespace());
+            }
+
+            return super.loadTable(identifier);
+          }
+        };
+    catalog.initialize("catalog", Map.of());
+    Namespace namespace = Namespace.of("cached_namespace");
+    TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    // The record processor asks the cache whether the table exists before it calls the updater, so
+    // the
+    // updater is given the cached answer, which does not carry the exception of the lookup.
+    Tuple2<Boolean, Exception> first = cache.exists(tableIdentifier);
+    assertThat(first.f0).isFalse();
+    assertThat(first.f1).isInstanceOf(NoSuchNamespaceException.class);
+    assertThat(cache.exists(tableIdentifier).f1).isNull();
+
+    tableUpdater.update(
+        tableIdentifier,
+        SnapshotRef.MAIN_BRANCH,
+        SCHEMA,
+        PartitionSpec.unpartitioned(),
+        TableCreator.DEFAULT);
+
+    assertThat(catalog.namespaceExists(namespace)).isTrue();
+    assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+  }
+
+  private static Catalog catalogWithoutNamespaceSupport(Supplier<RuntimeException> lookupFailure) {
+    return new Catalog() {
+      @Override
+      public List<TableIdentifier> listTables(Namespace namespace) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public Table loadTable(TableIdentifier identifier) {
+        throw lookupFailure.get();
+      }
+
+      @Override
+      public boolean dropTable(TableIdentifier identifier, boolean purge) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public void renameTable(TableIdentifier from, TableIdentifier to) {
+        throw new UnsupportedOperationException();
+      }
+    };
+  }
+
+  @Test
+  void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() {
+    Catalog catalog =
+        catalogWithoutNamespaceSupport(
+            () -> new NoSuchNamespaceException("Namespace does not exist"));
+    TableIdentifier tableIdentifier =
+        TableIdentifier.of(Namespace.of("missing_namespace"), "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageContaining("does not support namespaces")
+        .hasCauseInstanceOf(NoSuchNamespaceException.class);
+  }
+
+  @Test
+  void testCatalogWithoutNamespaceSupportFailsWhateverTheLookupReported() {
+    Catalog catalog =
+        catalogWithoutNamespaceSupport(() -> new NoSuchTableException("Table does not exist"));
+    TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("some_namespace"), "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    // The first answer carries the exception of the lookup, the second one comes from the cache and
+    // carries none: both must fail the same way.
+    assertThat(cache.exists(tableIdentifier).f1).isInstanceOf(NoSuchTableException.class);
+    assertThat(cache.exists(tableIdentifier).f1).isNull();
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessageContaining("does not support namespaces");
+  }
+
+  @Test
+  void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() {
+    Catalog catalog = CATALOG_EXTENSION.catalog();
+    TableIdentifier tableIdentifier = TableIdentifier.parse("myTable");
+    RuntimeException failure = new RuntimeException("catalog unavailable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS) {
+          @Override
+          Tuple2<Boolean, Exception> exists(TableIdentifier identifier) {
+            return Tuple2.of(false, failure);
+          }
+        };
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThatThrownBy(
+            () ->
+                tableUpdater.update(
+                    tableIdentifier,
+                    SnapshotRef.MAIN_BRANCH,
+                    SCHEMA,
+                    PartitionSpec.unpartitioned(),
+                    TableCreator.DEFAULT))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessageContaining("Unexpected failure looking up table")
+        .hasCause(failure);
+    assertThat(catalog.tableExists(tableIdentifier)).isFalse();
+  }
+
   @Test
   void testSpecCreation() {
     Catalog catalog = CATALOG_EXTENSION.catalog();
@@ -453,4 +600,42 @@ void testTableCreationWithExistingNamespace() {
     assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
     assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME);
   }
+
+  @Test
+  void testTableCreationWhenCatalogReportsMissingNamespaceAsNoSuchTable() {
+    // The Glue catalog answers a lookup in a database that does not exist with
+    // NoSuchTableException,
+    // not NoSuchNamespaceException, and then refuses to create the table.
+    InMemoryCatalog catalog =
+        new InMemoryCatalog() {
+          @Override
+          public Table loadTable(TableIdentifier identifier) {
+            if (!namespaceExists(identifier.namespace())) {
+              throw new NoSuchTableException("Table does not exist: %s", identifier);
+            }
+
+            return super.loadTable(identifier);
+          }
+        };
+    catalog.initialize("catalog", Map.of());
+    Namespace namespace = Namespace.of("glue_like_namespace");
+    TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+    TableMetadataCache cache =
+        new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+    TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+    assertThat(catalog.namespaceExists(namespace)).isFalse();
+
+    Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+        tableUpdater.update(
+            tableIdentifier,
+            SnapshotRef.MAIN_BRANCH,
+            SCHEMA,
+            PartitionSpec.unpartitioned(),
+            TableCreator.DEFAULT);
+
+    assertThat(catalog.namespaceExists(namespace)).isTrue();
+    assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+    assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
+  }
 }