Repository: apache/flink
File: flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/internal/StreamTableEnvironmentImpl.java (Java)

Code before:
                FlinkUserCodeClassLoaders.create(
                        new URL[0], settings.getUserClassLoader(), settings.getConfiguration());
        final Executor executor = lookupExecutor(userClassLoader, executionEnvironment);

        final TableConfig tableConfig = TableConfig.getDefault();
        tableConfig.setRootConfiguration(executor.getConfiguration());
        tableConfig.addConfiguration(settings.getConfiguration());

        final ResourceManager resourceManager =
                new ResourceManager(settings.getConfiguration(), userClassLoader);
        final ModuleManager moduleManager = new ModuleManager();

        final CatalogStore catalogStore;
        final CatalogStoreFactory catalogStoreFactory;
        if (settings.getCatalogStore().isPresent()) {
            catalogStore = settings.getCatalogStore().get();
            catalogStoreFactory = null;
        } else {
            catalogStoreFactory =
                    ApiFactoryUtil.findAndCreateCatalogStoreFactory(
                            settings.getConfiguration(), userClassLoader);
            final CatalogStoreFactory.Context catalogStoreFactoryContext =
                    ApiFactoryUtil.buildCatalogStoreFactoryContext(
                            settings.getConfiguration(), userClassLoader);
            catalogStoreFactory.open(catalogStoreFactoryContext);
            catalogStore = catalogStoreFactory.createCatalogStore();
        }

        final CatalogManager catalogManager =
                CatalogManager.newBuilder()
                        .classLoader(userClassLoader)
                        .config(tableConfig)
                        .defaultCatalog(
                                settings.getBuiltInCatalogName(),
                                new GenericInMemoryCatalog(
                                        settings.getBuiltInCatalogName(),
                                        settings.getBuiltInDatabaseName()))
                        .executionConfig(executionEnvironment.getConfig())
                        .catalogModificationListeners(
                                TableFactoryUtil.findCatalogModificationListenerList(
                                        settings.getConfiguration(), userClassLoader))
                        .catalogStoreHolder(
                                CatalogStoreHolder.newBuilder()
                                        .classloader(userClassLoader)
                                        .config(tableConfig)
                                        .catalogStore(catalogStore)
                                        .factory(catalogStoreFactory)
                                        .build())
                        .build();

        final FunctionCatalog functionCatalog =

Review comment (refactor):
Can we move this code also to `ApiFactoryUtil`? In the end the signature can be `ApiFactoryUtil.getOrCreateCatalogStore(@Nullable CatalogStore providedCatalogStore)`. This avoids duplicate code in `TableEnvironmentImpl.java‎`

Code after:
                FlinkUserCodeClassLoaders.create(
                        new URL[0], settings.getUserClassLoader(), settings.getConfiguration());
        final Executor executor = lookupExecutor(userClassLoader, executionEnvironment);

        final TableConfig tableConfig = TableConfig.getDefault();
        tableConfig.setRootConfiguration(executor.getConfiguration());
        tableConfig.addConfiguration(settings.getConfiguration());

        final ResourceManager resourceManager =
                new ResourceManager(settings.getConfiguration(), userClassLoader);
        final ModuleManager moduleManager = new ModuleManager();

        final ApiFactoryUtil.CatalogStoreResult catalogStoreResult =
                ApiFactoryUtil.getOrCreateCatalogStore(
                        settings.getCatalogStore(), settings.getConfiguration(), userClassLoader);
        final CatalogStore catalogStore = catalogStoreResult.getCatalogStore();
        final CatalogStoreFactory catalogStoreFactory = catalogStoreResult.getCatalogStoreFactory();

        final CatalogManager catalogManager =
                CatalogManager.newBuilder()
                        .classLoader(userClassLoader)
                        .config(tableConfig)
                        .defaultCatalog(
                                settings.getBuiltInCatalogName(),
                                new GenericInMemoryCatalog(
                                        settings.getBuiltInCatalogName(),
                                        settings.getBuiltInDatabaseName()))
                        .executionConfig(executionEnvironment.getConfig())
                        .catalogModificationListeners(
                                TableFactoryUtil.findCatalogModificationListenerList(
                                        settings.getConfiguration(), userClassLoader))
                        .catalogStoreHolder(
                                CatalogStoreHolder.newBuilder()
                                        .classloader(userClassLoader)
                                        .config(tableConfig)
                                        .catalogStore(catalogStore)
                                        .factory(catalogStoreFactory)
                                        .build())
                        .build();

        final FunctionCatalog functionCatalog =
                new FunctionCatalog(tableConfig, resourceManager, catalogManager, moduleManager);

        final Planner planner =
                PlannerFactoryUtil.createPlanner(
                        executor,
                        tableConfig,
                        userClassLoader,
                        moduleManager,
                        catalogManager,
                        functionCatalog);