Repository: apache/flink
File: flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/module/StateModule.java (Java)

Code before:
import java.util.function.Function;
import java.util.stream.Collectors;

/** Module of state in Flink. */
@Experimental
public class StateModule implements Module {

    public static final String IDENTIFIER = "state";

    public static final StateModule INSTANCE = new StateModule();

    private final Map<String, BuiltInFunctionDefinition> normalizedFunctions;
    private final Set<String> functionNamesWithInternal;
    private final Set<String> functionNamesWithoutInternal;

    private StateModule() {
        final List<BuiltInFunctionDefinition> definitions = new ArrayList<>();

        definitions.add(SavepointMetadataTableFunction.SAVEPOINT_METADATA);
        ServiceLoader.load(DynamicBuiltInFunctionDefinitionFactory.class)
                .iterator()
                .forEachRemaining(
                        f -> {
                            if (f.factoryIdentifier().startsWith(IDENTIFIER)) {
                                definitions.addAll(f.getBuiltInFunctionDefinitions());
                            }
                        });

        this.normalizedFunctions =
                definitions.stream()
                        .collect(
                                Collectors.toMap(
                                        f -> f.getName().toUpperCase(Locale.ROOT),
                                        Function.identity()));
        this.functionNamesWithInternal =
                definitions.stream()
                        .map(BuiltInFunctionDefinition::getName)
                        .collect(Collectors.toSet());
        this.functionNamesWithoutInternal =
                definitions.stream()
                        .filter(f -> !f.isInternal())
                        .map(BuiltInFunctionDefinition::getName)
                        .collect(Collectors.toSet());
    }

    @Override
    public Set<String> listFunctions() {
        return listFunctions(false);
    }

    @Override

Review comment (refactor):
maybe i misunderstood something, but what if two plugins register the same function name? if two providers define get_offsets, which one wins? so maybe after building, we can consider `Collections.unmodifiableList(definitions)` (or deduplicate names before normalization) to avoid accidental mutation. wdyt?

Code after:
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

/** Module of state in Flink. */
@Experimental
public class StateModule implements Module {

    private static final Logger LOG = LoggerFactory.getLogger(StateModule.class);

    public static final String IDENTIFIER = "state";

    public static final StateModule INSTANCE = new StateModule();

    private final Map<String, BuiltInFunctionDefinition> normalizedFunctions;
    private final Set<String> functionNamesWithInternal;
    private final Set<String> functionNamesWithoutInternal;

    private StateModule() {
        final List<BuiltInFunctionDefinition> definitions = new ArrayList<>();

        definitions.add(SavepointMetadataTableFunction.SAVEPOINT_METADATA);
        ServiceLoader.load(DynamicBuiltInFunctionDefinitionFactory.class)
                .iterator()
                .forEachRemaining(
                        f -> {
                            if (f.factoryIdentifier().startsWith(IDENTIFIER + ".")) {
                                definitions.addAll(f.getBuiltInFunctionDefinitions());
                            }
                        });
        checkDuplicatedFunctions(definitions);

        this.normalizedFunctions =
                definitions.stream()
                        .collect(
                                Collectors.toMap(
                                        f -> f.getName().toUpperCase(Locale.ROOT),
                                        Function.identity()));
        this.functionNamesWithInternal =
                definitions.stream()
                        .map(BuiltInFunctionDefinition::getName)
                        .collect(Collectors.toSet());
        this.functionNamesWithoutInternal =
                definitions.stream()
                        .filter(f -> !f.isInternal())
                        .map(BuiltInFunctionDefinition::getName)
                        .collect(Collectors.toSet());