You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "vishnuprakaz (via GitHub)" <gi...@apache.org> on 2026/07/30 17:47:52 UTC
[PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
vishnuprakaz opened a new pull request, #17437:
URL: https://github.com/apache/iceberg/pull/17437
The dynamic sink's `TableSerializerCache` loads a catalog with `CatalogLoader.loadCatalog()` on every cache miss (an unknown schema or spec id, or an LRU eviction) but never closes it. Each call builds a fresh catalog whose resources, such as a REST HTTP client and thread pool or a Hive metastore client pool, are released only on close, so a long-running job that keeps missing the cache leaks connections and threads.
The cache lives inside a Flink `TypeSerializer` that has no teardown hook, so it now closes the catalog right after reading the table's schemas and specs. This matches how `TableLoader` closes the catalogs it opens.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "vishnuprakaz (via GitHub)" <gi...@apache.org>.
vishnuprakaz commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3729035765
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java:
##########
@@ -120,9 +126,18 @@ private class SerializerInfo {
}
private void update() {
- Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
- schemas = table.schemas();
- specs = table.specs();
+ // The serializer has no teardown hook, so a catalog cannot be held for reuse; load and
+ // close one per cache miss.
+ try (TableLoader tableLoader =
+ TableLoader.fromCatalog(catalogLoader, TableIdentifier.parse(tableName))) {
+ tableLoader.open();
+ Table table = tableLoader.loadTable();
+ schemas = table.schemas();
+ specs = table.specs();
+ } catch (IOException e) {
+ // only close() throws IOException here; a failed close should not fail the lookup
+ LOG.warn("Failed to close catalog for table {}", tableName, e);
+ }
Review Comment:
That makes a lot of sense, loading once and reusing is better trade actually. I was thinking per table and missed how many tables the dynamic sink could be routing. Made the changes, happy to adjust if anything looks off :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3740071321
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java:
##########
@@ -49,6 +50,9 @@
class TableSerializerCache implements Serializable {
private final CatalogLoader catalogLoader;
+
+ // Intentionally not closed; the catalog is reused for the serializer's lifetime.
Review Comment:
`transient` field should be next to the other transient field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3710450832
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java:
##########
@@ -121,4 +127,25 @@ void testCacheSize() {
cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000);
assertThat(cache.maximumSize()).isEqualTo(1000);
}
+
+ @Test
+ void testClosesCatalogAfterSchemaLookup() throws Exception {
+ Table table =
+ CATALOG_EXTENSION
+ .catalogLoader()
+ .loadCatalog()
+ .createTable(TableIdentifier.of("table"), schema1);
+
+ Catalog catalog = mock(Catalog.class, withSettings().extraInterfaces(Closeable.class));
+ when(catalog.loadTable(any(TableIdentifier.class))).thenReturn(table);
+ CatalogLoader catalogLoader = mock(CatalogLoader.class);
+ when(catalogLoader.loadCatalog()).thenReturn(catalog);
+ cache = new TableSerializerCache(catalogLoader, 10);
Review Comment:
Could we write this test without using Mockito?
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java:
##########
@@ -120,9 +127,22 @@ private class SerializerInfo {
}
private void update() {
- Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
- schemas = table.schemas();
- specs = table.specs();
+ // The serializer has no teardown hook, so the freshly loaded catalog is closed here, after
+ // reading the table metadata, to avoid leaking one per cache miss.
+ Catalog catalog = catalogLoader.loadCatalog();
+ try {
+ Table table = catalog.loadTable(TableIdentifier.parse(tableName));
+ schemas = table.schemas();
+ specs = table.specs();
+ } finally {
+ if (catalog instanceof Closeable) {
+ try {
+ ((Closeable) catalog).close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close catalog {}", catalog.name(), e);
+ }
+ }
Review Comment:
I think we want to move to a TaskManager-level catalog cache that we close on job shutdown. Out of scope for this PR. I think @swapna267 is looking into that already.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Reuse the catalog in the dynamic sink serializer cache [iceberg]
Posted by "vishnuprakaz (via GitHub)" <gi...@apache.org>.
vishnuprakaz commented on PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#issuecomment-5225770952
Thanks for all the reviews on this one @mxm learned a lot. Appreciate the patience!.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "vishnuprakaz (via GitHub)" <gi...@apache.org>.
vishnuprakaz commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3711607681
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java:
##########
@@ -120,9 +127,22 @@ private class SerializerInfo {
}
private void update() {
- Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
- schemas = table.schemas();
- specs = table.specs();
+ // The serializer has no teardown hook, so the freshly loaded catalog is closed here, after
+ // reading the table metadata, to avoid leaking one per cache miss.
+ Catalog catalog = catalogLoader.loadCatalog();
+ try {
+ Table table = catalog.loadTable(TableIdentifier.parse(tableName));
+ schemas = table.schemas();
+ specs = table.specs();
+ } finally {
+ if (catalog instanceof Closeable) {
+ try {
+ ((Closeable) catalog).close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close catalog {}", catalog.name(), e);
+ }
+ }
Review Comment:
yeah, makes sense it's a better approach that would remove the need for per lookup. Happy to keep this scoped for now. Thanks for letting me know!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Reuse the catalog in the dynamic sink serializer cache [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#issuecomment-5239220253
@vishnuprakaz Could you open a new PR with a backport for Flink versions 2.0 and 1.20?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Reuse the catalog in the dynamic sink serializer cache [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm merged PR #17437:
URL: https://github.com/apache/iceberg/pull/17437
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Reuse the catalog in the dynamic sink serializer cache [iceberg]
Posted by "vishnuprakaz (via GitHub)" <gi...@apache.org>.
vishnuprakaz commented on PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#issuecomment-5239587978
Thanks for merging @mxm! Sure will do a backport for Flink 2.0 and 1.20.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3718122762
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableSerializerCache.java:
##########
@@ -121,4 +125,48 @@ void testCacheSize() {
cache = new TableSerializerCache(CATALOG_EXTENSION.catalogLoader(), 1000);
assertThat(cache.maximumSize()).isEqualTo(1000);
}
+
+ @Test
+ void testClosesCatalogAfterSchemaLookup() {
+ CloseCountingInMemoryCatalog catalog = new CloseCountingInMemoryCatalog();
+ catalog.initialize("tracking", Map.of());
+ catalog.createNamespace(Namespace.of("db"));
+ Table table = catalog.createTable(TableIdentifier.of("db", "table"), schema1);
+ cache = new TableSerializerCache(new SingleCatalogLoader(catalog), 10);
+
+ // schema/spec ids are unknown, so this misses the cache and loads a catalog to resolve them
+ cache.serializerWithSchemaAndSpec(
+ "db.table", table.schema().schemaId(), PartitionSpec.unpartitioned().specId());
+
+ assertThat(catalog.closeCount).isEqualTo(1);
+ }
+
+ private static class CloseCountingInMemoryCatalog extends InMemoryCatalog {
+ private int closeCount = 0;
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ closeCount += 1;
+ }
+ }
+
+ private static class SingleCatalogLoader implements CatalogLoader {
Review Comment:
Can we replace this with `HadoopCatalogExtension`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3719178672
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java:
##########
@@ -120,9 +127,22 @@ private class SerializerInfo {
}
private void update() {
- Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
- schemas = table.schemas();
- specs = table.specs();
+ // The serializer has no teardown hook, so the freshly loaded catalog is closed here, after
+ // reading the table metadata, to avoid leaking one per cache miss.
+ Catalog catalog = catalogLoader.loadCatalog();
+ try {
+ Table table = catalog.loadTable(TableIdentifier.parse(tableName));
+ schemas = table.schemas();
+ specs = table.specs();
+ } finally {
+ if (catalog instanceof Closeable) {
+ try {
+ ((Closeable) catalog).close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close catalog {}", catalog.name(), e);
+ }
+ }
Review Comment:
Would it be feasible to let the `CatalogLoader` manage the lifecycle of the catalog and reuse it when necessary? It is a bit odd to check for `Closable` when the `Catalog` interface doesn't contain this interface.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "vishnuprakaz (via GitHub)" <gi...@apache.org>.
vishnuprakaz commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3719780763
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java:
##########
@@ -120,9 +127,22 @@ private class SerializerInfo {
}
private void update() {
- Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
- schemas = table.schemas();
- specs = table.specs();
+ // The serializer has no teardown hook, so the freshly loaded catalog is closed here, after
+ // reading the table metadata, to avoid leaking one per cache miss.
+ Catalog catalog = catalogLoader.loadCatalog();
+ try {
+ Table table = catalog.loadTable(TableIdentifier.parse(tableName));
+ schemas = table.schemas();
+ specs = table.specs();
+ } finally {
+ if (catalog instanceof Closeable) {
+ try {
+ ((Closeable) catalog).close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close catalog {}", catalog.name(), e);
+ }
+ }
Review Comment:
On this I looked into having the loader manage and reuse the catalog. the problem I ran into is that the serializer has no teardown hook, so as far as I can tell nothing at this layer could ever close a held catalog. Does that match your thinking that the reuse part belongs in the TaskManager level cache?
While checking, I noticed loadCatalog() has six other call sites in the sink that don't cloae the catalog either. The operator-level ones do have close() hooks, would it be worth a similar interim fix for those, or better to leave them for the TaskManager level cache work?
For the Closeable check here.... would it make sense to switch update() to TableLoader.fromCatalog() in a try with resources, like FlinkSink and IcebergSink do for one shot loads?
but the lifecycle (and the Closeable check) would live inside TableLoader instead of this class.
Happy to make that change if it sounds right : ).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "vishnuprakaz (via GitHub)" <gi...@apache.org>.
vishnuprakaz commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3719780763
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java:
##########
@@ -120,9 +127,22 @@ private class SerializerInfo {
}
private void update() {
- Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
- schemas = table.schemas();
- specs = table.specs();
+ // The serializer has no teardown hook, so the freshly loaded catalog is closed here, after
+ // reading the table metadata, to avoid leaking one per cache miss.
+ Catalog catalog = catalogLoader.loadCatalog();
+ try {
+ Table table = catalog.loadTable(TableIdentifier.parse(tableName));
+ schemas = table.schemas();
+ specs = table.specs();
+ } finally {
+ if (catalog instanceof Closeable) {
+ try {
+ ((Closeable) catalog).close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close catalog {}", catalog.name(), e);
+ }
+ }
Review Comment:
On this I looked into having the loader manage and reuse the catalog. the problem I ran into is that the serializer has no teardown hook, so as far as I can tell nothing at this layer could ever close a held catalog. Does that match your thinking that the reuse part belongs in the TaskManager level cache?
While checking, I noticed loadCatalog() has six other call sites in the sink that don't cloae the catalog either. The operator-level ones do have close() hooks, would it be worth a similar interim fix for those, or better to leave them for the TaskManager level cache work?
For the Closeable check here.... would it make sense to switch update() to TableLoader.fromCatalog() in a try with resources, like FlinkSink and IcebergSink do for one shot loads?
but the lifecycle (and the Closeable check) would live inside TableLoader instead of this class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Close catalog opened by the dynamic sink serializer cache [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17437:
URL: https://github.com/apache/iceberg/pull/17437#discussion_r3727454796
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableSerializerCache.java:
##########
@@ -120,9 +126,18 @@ private class SerializerInfo {
}
private void update() {
- Table table = catalogLoader.loadCatalog().loadTable(TableIdentifier.parse(tableName));
- schemas = table.schemas();
- specs = table.specs();
+ // The serializer has no teardown hook, so a catalog cannot be held for reuse; load and
+ // close one per cache miss.
+ try (TableLoader tableLoader =
+ TableLoader.fromCatalog(catalogLoader, TableIdentifier.parse(tableName))) {
+ tableLoader.open();
+ Table table = tableLoader.loadTable();
+ schemas = table.schemas();
+ specs = table.specs();
+ } catch (IOException e) {
+ // only close() throws IOException here; a failed close should not fail the lookup
+ LOG.warn("Failed to close catalog for table {}", tableName, e);
+ }
Review Comment:
I'm not sure loading / closing on every update is a good idea. Can we store `Catalog` in a field? This will get rid of the duplicate loading and only load once per job.
If we are concerned about not closing one instance, we could close it via `RuntimeContext#registerUserCodeClassLoaderReleaseHookIfAbsent`, but I'm not sure this is necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org