Kafka Connect: Handle AccessDeniedException in auto-create

[yadavay-amzn]

Summary

Fixes #13758.

When iceberg.tables.auto-create-enabled is set, the Kafka Connect sink connector calls GlueCatalog.createNamespace() which can throw software.amazon.awssdk.services.glue.model.AccessDeniedException if the user lacks glue:CreateDatabase permission. This exception was not caught, crashing the connector even when the database already exists.

Changes

Root cause fix — GlueCatalog.createNamespace():

• Catch AccessDeniedException and wrap it as Iceberg's ForbiddenException, consistent with how GlueTableOperations already handles this exception for table operations.

Defense-in-depth — IcebergWriterFactory.createNamespaceIfNotExist():

• Add NotAuthorizedException to the existing catch block alongside AlreadyExistsException and ForbiddenException, so auth exceptions from any catalog implementation are handled gracefully.

Testing

• Added testCreateNamespaceAccessDenied in TestGlueCatalog: verifies AccessDeniedException is wrapped as ForbiddenException.
• Added testCreateNamespaceHandlesForbiddenException and testCreateNamespaceHandlesNotAuthorizedException in TestIcebergWriterFactory: verifies both exception types are swallowed during namespace creation.

[yadavay-amzn]

@Baunsgaard @nastra Could you take a look when you get a chance? This handles AccessDeniedException gracefully in the Kafka Connect auto-create path when the user lacks glue:CreateDatabase permission. Fixed the PR title format as well.

[cloventt]

Wouldn't this approach still cause the connector to crash if glue.CreateDatabase permission does not exist? I don't think this solves the problem. Ideally we want the connector to proceed if the database already exists.

There is another approach that would work. The outer method is called createNamespaceIfNotExist(), which to me implies that we should first test to see if the namespace exists, and if it not, create it. We could call glue.getDatabase() first, and then only if we receive a EntityNotFoundException then we proceed to call glue.createDatabase(). This would mean that if the Glue database already exists the connector would proceed without ever attempting to invoke glue.createDatabase.

[yadavay-amzn]

@cloventt Good suggestion. Updated to check namespaceExists() first and only call createNamespace() if the namespace does not exist. This way users with read-only Glue access never hit the permission error when the database already exists. The catch block is kept as defense-in-depth for concurrent creation or other edge cases. Also added a test verifying that createNamespace is never called when the namespace already exists.

[cloventt]

Nice one, I submitted my own PR so the maintainers can choose whichever impl they prefer #16297.

[Baunsgaard]

TBH, my feeling is that it should fail on these exception types, and only the AlreadyExistsException should be allowed to skip.

[Baunsgaard]

This seems wrong to me. We should at minimum do a LOG.warning() to highlight that we have an conflict. I do see the old code did not do any logging, but if we add it we can get some statistics on how often this error is encountered.

[yadavay-amzn]

@Baunsgaard Makes sense. Updated: the catch now only handles AlreadyExistsException (concurrent creation race) with a LOG.warn. Auth exceptions propagate so the user sees the real permission error.