yunfengzhou-hub commented on code in PR #10833:
URL: https://github.com/apache/paimon/pull/10833#discussion_r4409820909
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java:
##########
@@ -973,10 +1002,153 @@ public final void open() throws CatalogException {}
@Override
public final void close() throws CatalogException {
+ Throwable failure = null;
try {
catalog.close();
} catch (Exception e) {
- throw new CatalogException("Failed to close catalog " +
catalog.toString(), e);
+ failure = new CatalogException("Failed to close catalog " +
catalog.toString(), e);
+ }
+ try {
+ closeLakeStreamStoreCatalog();
+ } catch (Throwable t) {
+ failure = ExceptionUtils.firstOrSuppressed(t, failure);
+ }
+ if (failure != null) {
+ ExceptionUtils.rethrow(failure);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Lake-stream mode (PIP-39)
+ // ------------------------------------------------------------------------
+
+ /** Whether this catalog names a Lakestream Store. */
+ public boolean lakeStreamStoreConfigured() {
+ return lakeStreamStore() != null;
+ }
+
+ /** The configured store identifier, or {@code <unset>} when there is
none. */
+ public String lakeStreamStoreIdentifier() {
+ String store = lakeStreamStore();
+ return store == null ? "<unset>" : store;
+ }
+
+ /** The trimmed {@code lakestream.store} value, or {@code null} when it is
absent or blank. */
+ @Nullable
+ private String lakeStreamStore() {
+ String store = options.get(CatalogOptions.LAKE_STREAM_STORE);
+ return StringUtils.isNullOrWhitespaceOnly(store) ? null : store.trim();
+ }
+
+ /**
+ * Lazily discovers, creates, opens and caches the Lakestream Store
catalog. Lazy so that an
+ * unreachable or absent store does not stop this catalog from resolving
table metadata.
+ */
+ public synchronized org.apache.flink.table.catalog.Catalog
getOrCreateLakeStreamStoreCatalog() {
+ org.apache.flink.table.catalog.Catalog cached = lakeStreamStoreCatalog;
+ if (cached != null) {
+ return cached;
+ }
+
+ String identifier = lakeStreamStore();
+ if (identifier == null) {
+ throw new CatalogException(
+ String.format(
+ "Lake-stream mode requires the catalog option '%s'
to name the "
+ + "Lakestream Store implementation (e.g.
'%s' = 'fluss'), but "
+ + "catalog '%s' does not set it to a store
identifier.",
+ CatalogOptions.LAKE_STREAM_STORE.key(),
+ CatalogOptions.LAKE_STREAM_STORE.key(),
+ name));
+ }
+
+ CatalogFactory catalogFactory;
+ try {
+ catalogFactory =
+ FactoryUtil.discoverFactory(
+ Thread.currentThread().getContextClassLoader(),
Review Comment:
Fixed. FlinkCatalog now retains the class loader
FlinkCatalogFactory#createCatalog was given and uses it for both the SPI lookup
and the delegated catalog context.
Covered by testStoreDiscoveryUsesTheClassLoaderSuppliedToTheCatalogFactory,
which sets the thread context loader to one that cannot see the store's service
registration while the supplied loader can.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
