Kafka-connect: Handle namespace creation for auto table creation #10186
Conversation
| @@ -47,7 +54,7 @@ public class IcebergWriterFactoryTest { | |||
| @ValueSource(booleans = {true, false}) | |||
| @SuppressWarnings("unchecked") | |||
| public void testAutoCreateTable(boolean partitioned) { | |||
| Catalog catalog = mock(Catalog.class); | |||
| Catalog catalog = mock(InMemoryCatalog.class); | |||
There was a problem hiding this comment.
Need to change it to some catalog that supports namespace creation. As the code now calls createNamespaceIfNotExist
There was a problem hiding this comment.
I'd rather not rely on InMemoryCatalog API here. You could do something like this instead, Catalog catalog = mock(Catalog.class, withSettings().extraInterfaces(SupportsNamespaces.class));
...ct/kafka-connect/src/test/java/org/apache/iceberg/connect/data/IcebergWriterFactoryTest.java
Outdated
Show resolved
Hide resolved
| Namespace namespace = Namespace.of(Arrays.copyOfRange(levels, 0, index + 1)); | ||
| try { | ||
| ((SupportsNamespaces) catalog).createNamespace(namespace); | ||
| } catch (AlreadyExistsException ex) { |
There was a problem hiding this comment.
We went the other way with Flink: #7795
Can we also swallow the exception where the user doesn't have permission to create a namespace.
There was a problem hiding this comment.
I assumed if the auto create table is enabled, means user has permission to create tables. So, They can create Namespaces also. But that assumptions can be wrong for some catalogs. Let me catch ForbiddenException also for ignoring it.
There was a problem hiding this comment.
Thanks for working on this @ajantha-bhat 👍
Any concerns @bryanck? We could also make this configurable through a flag, but I don't think we want to overcomplicate things.
| for (int index = 0; index < levels.length; index++) { | ||
| Namespace namespace = Namespace.of(Arrays.copyOfRange(levels, 0, index + 1)); | ||
| try { | ||
| ((SupportsNamespaces) catalog).createNamespace(namespace); |
There was a problem hiding this comment.
This can cause a ClassCastException so we should check that the catalog implements SupportsNamespaces first.
| TableIdentifier.of(Namespace.of("foo1", "foo2", "foo3"), "bar"); | ||
| Schema schema = new Schema(Types.NestedField.required(1, "id", Types.StringType.get())); | ||
|
|
||
| try (InMemoryCatalog catalog = new InMemoryCatalog()) { |
There was a problem hiding this comment.
I'd prefer using the ArgumentCaptor approach, rather than relying on InMemoryCatalog. This is consistent with the auto create test.
There was a problem hiding this comment.
done. Updated the existing testcase
|
ping @bryanck |
|
Thanks for the contribution @ajantha-bhat , it looks good! |
Some catalogs like Inmemory, Hive, Nessie, REST catalog expects the namespace to be present before creating the table. Hence, the auto table creation feature will not work currently for these catalogs if the namespace is missing.
Added a support to create the missing namespace and its parents while auto creating the table.