EventStreamCatalogConfigurationFactory.java

package org.wikimedia.eventutilities.flink.table.catalog;

import java.util.Collection;
import java.util.Collections;
import java.util.Map;

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.wikimedia.eventutilities.core.event.EventStreamFactory;

import lombok.Builder;
import lombok.NonNull;

@Builder(toBuilder = true)
public class EventStreamCatalogConfigurationFactory {
    @Builder.Default
    private final Map<String, String> catalogOptions = Collections.emptyMap();
    @Builder.Default
    private final Collection<ConfigOption<?>> pseudoTableOptions = Collections.emptyList();
    @Builder.Default
    private final Collection<ConfigOption<?>> overrideOptions = Collections.emptyList();
    @Builder.Default
    private final Collection<ConfigOption<?>> sharedOptions = Collections.emptyList();
    @Builder.Default
    private final Map<String, String> defaultTableOptions = Collections.emptyMap();
    @Builder.Default
    private final Map<String, String> defaultKafkaOptions = Collections.emptyMap();
    @Builder.Default
    private final Map<String, String> defaultFormatOptions = Collections.emptyMap();

    public EventStreamCatalogConfiguration createTableEventConfiguration(
        @NonNull Map<String, String> tableOptions,
        @NonNull EventStreamFactory eventStreamFactory
    ) {
        return new CreateTableEventStreamCatalogConfiguration(
            Configuration.fromMap(this.catalogOptions),
            Configuration.fromMap(tableOptions),
            this.pseudoTableOptions,
            this.overrideOptions,
            this.sharedOptions,
            this.defaultTableOptions,
            this.defaultKafkaOptions,
            this.defaultFormatOptions,
            eventStreamFactory
        );
    }

    public EventStreamCatalogConfiguration createTableLikeEventConfiguration(
        @NonNull Map<String, String> tableOptions,
        @NonNull EventStreamFactory eventStreamFactory
    ) {
        return new CreateTableLikeEventStreamCatalogConfiguration(
            Configuration.fromMap(this.catalogOptions),
            Configuration.fromMap(tableOptions),
            this.pseudoTableOptions,
            this.overrideOptions,
            this.sharedOptions,
            this.defaultTableOptions,
            this.defaultKafkaOptions,
            this.defaultFormatOptions,
            eventStreamFactory
        );
    }
}