/**
 * Different events will be fired when a catalog/database/table is modified. The customized listener can get and report specific information from the event according to the event type.
 */
@PublicEvolving
public interface CatalogEventListener {
    /* Event fired after a catalog/database/table is modified. */
    void onEvent(CatalogEvent catalogEvent);

    /* The basic class for catalog related event. */
    @PublicEvolving
    public abstract class CatalogEvent {
        /* The catalog of the event. */
        Catalog catalog();
        /* The name of catalog. */
        String catalogName();
    }

    /* The basic class for database related event. */
    public abstract class BaseDatabaseEvent extends CatalogEvent {
        String databaseName();  
    }

    /* Event for database creation. */
    @PublicEvolving
    public class CreateDatabaseEvent extends BaseDatabaseEvent {
        CatalogDatabase database();
        boolean ignoreIfExists();
    }

    /* Event for alter database. */
    @PublicEvolving
    public class AlterDatabaseEvent extends BaseDatabaseEvent {
        CatalogDatabase newDatabase();
        boolean ignoreIfNotExists();
    }

    /* Event for dropping database. */
    @PublicEvolving
    public class DropDatabaseEvent extends BaseDatabaseEvent {
        boolean ignoreIfExists();
    }

    /* Base table event, provides column list, primary keys, partition keys, watermarks and properties in CatalogBaseTable. The table can be source or sink. */
    public abstract class BaseTableEvent extends CatalogEvent {
        ObjectIdentifier identifier();  
        CatalogBaseTable table();
    }

    /* Event for table creation. */
    @PublicEvolving
    public class CreateTableEvent extends BaseTableEvent {
        boolean ignoreIfExists();
    }

    /* Event for altering table, provides all information in old table and new table. */
    @PublicEvolving
    public class AlterTableEvent extends BaseTableEvent {
        List<TableChange> tableChanges();
        boolean ignoreIfExists();
    }

    /* Event for dropping table. */
    @PublicEvolving
    public class DropTableEvent extends BaseTableEvent {
        boolean ignoreIfExists();   
    }
}

/* Factory for catalog listener. */
@PublicEvolving
public interface CatalogEventListenerFactory {
    public CatalogEventListener createListener(Configuration configuration, ClassLoader classLoader);
}

/* Add listeners in the catalog context. */
@PublicEvolving
public interface CatalogFactory {
    /** Add listeners in the context. */
    @PublicEvolving
    interface Context {
        /* Get the listeners from context if they are exists. */
        List<CatalogEventListener> listeners();
    }
}

/* Storage identifier for different physical table. */
@PublicEvolving
public class StorageIdentifier {
    /* Storage type such as kafka, hive, iceberg or paimon. */
    String type();

    /* Identifier which identify the unique physical table. */
    String identifier();

    /* Properties for the storage identifier, users can get value from different keys for different storages, such as broker and topic for kafka. */
    Map<String, String> properties();
}

/* Storage identifier factory is loaded with specific connector type and create {@link StorageIdentifier}. */
@PublicEvolving
public interface StorageIdentifierFactory {
    /* Create storage identifier for different connector type. */
    StorageIdentifier createDynamicTableStorage(Configuration config, ClassLoader classLoader);
}

@PublicEvolving
public interface CatalogTable {
    /* Get physical storage identifier for the table. */
    StorageIdentifier storage();
}

/**
 * This listener will be notified before job is submitted in {@link RestClusterClient}.
 */
@PublicEvolving
public interface JobSubmissionListener {
    /* Event is fired before a job is submitted. */
    void onEvent(JobSubmissionEvent submissionEvent);

    /* Event for job submission. */
    @PublicEvolving
    public class JobSubmissionEvent {
        JobID jobId();
        String jobName();
        JobLogicalPlan plan();
    }
}

/* Factory for job submission listener. */
@PublicEvolving
public interface JobSubmissionListenerFactory {
    public JobSubmissionListener createListener(Configuration configuration, ClassLoader classLoader);
}

/**
 * Job logical plan is built according to StreamGraph in the client. Users can get sources, sinks and the relationship between nodes from plan.
 */
@PublicEvolvig
public interface JobLogicalPlan {
    /* Job type, TableOrSQL or DataStream. */
    String jobType();

    /* Job execution type, BATCH or STREAMING. */
    String executionType();

    /* Source lineage list. */
    List<JobSourceLineage> sources();

    /* Sink lineage list. */
    List<JobSinkLineage> sinks(); 

    /* Get sink table descriptor for table/sql job with given sink identifier, it will always get null for DataStream jobs. */
    SinkTableDescriptor sinkTable(String identifier);

    /* Job configuration. */
    Map<String, String> config(); 
}
 
/* Source info of the job plan. */
@PublicEvolving
public class JobSourceLineage {
    StorageIdentifier identifier();

    /* Source column name list. */
    List<String> columns();
}
 
/* Sink info of the job plan. */
@PublicEvolving 
public class JobSinkLineage {
    StorageIdentifier identifier();

    /* Sink column name list. */
    List<String> columns();

    /* Source column lineages, the key is the column in the sink and the value is the source columns list. */
    Map<String, List<SourceColumnLineage>> columnLineages();
}
 
/* Source column list for sink vertex. */
@PublicEvolving  
public class SourceColumnLineage {
    /* Source identifier. */
    String identifier();

    /* Source name list for the given sink. */
    List<String> columns();
}

@PublicEvolving
public class SinkTableDescriptor {
    /* Modify type, INSERT/UPDATE/DELETE. */
    String modifyType();

    /* Update mode, APPEND/RETRACT/UPSERT. */
    String updateMode();
    boolean overwrite();
}

/**
 * Add set source lineage method in data stream source.
 */
@Public
public class DataStreamSource {
    private JobSourceLineage lineage;

    public DataStreamSource setLineage(JobSourceLineage lineage);
}

/**
 * Add set sink lineage method in data stream sink.
 */
@Public
public class DataStreamSink {
    private JobSinkLineage lineage;

    public DataStreamSink setLineage(JobSinkLineage lineage);
}

/**
 * When job status is changed in job manager, it will generate job event and notify job execution listener.
 */
@PublicEvolving
public interface JobExecutionListener {
    /* Event fired after job status has been changed. */ 
    void onEvent(JobStatusEvent jobStatusEvent);

    /* Job status event with plan. */
    @PublicEvolving
    public class JobStatusEvent {
        JobID jobId();
        String jobName();
        JobStatus oldStatus();
        JobStatus newStatus();
        /* Exception for job when it is failed. */
        @Nullable Throwable exception();
    }
}

/* Factory for job execution listener. */
@PublicEvolving
public interface JobExecutionListenerFactory {
    public JobExecutionListener createListener(Configuration configuration, ClassLoader classLoader);
}

# Config catalog event listeners.
table.catalog.listeners: {job catalog listener factory1},{job catalog listener factory2}

# Existing config job submission listeners.
execution.job-submission-listeners: {job submission listener factory1},{job submission listener factory2}

# Config job execution listeners.
jobmanager.execution.listeners: {job execution listener factory1},{job execution listener factory2}

/* Listeners related operations in the catalog manager. */
public final class CatalogManager {
    /* Create catalog manager with listener list. */
    private CatalogManager(
            String defaultCatalogName,
            Catalog defaultCatalog,
            DataTypeFactory typeFactory,
            ManagedTableListener managedTableListener,
            List<CatalogEventListener> listeners);

    /* Notify the listeners with given catalog event. */
    private void notify(CatalogEvent event) {
        for (CatalogEventListener listener : listeners) {
            listener.onEvent(event);
        }
    }

    /* Notify listener for tables. */
    public void createTable/dropTable/alterTable(...) {
        ....;
        notify(Create Different Table Event);
    }

    /* Builder for catalog manager. */
    public static final class Builder {
        Builder listeners(List<CatalogEventListener> listeners);
    }
}

/* Listeners related operations in AbstractCatalog. */
public abstract class AbstractCatalog implements Catalog {
    /* Create the catalog with listeners. */
    public AbstractCatalog(String name, String defaultDatabase, List<CatalogEventListener> listeners); 

    /**
     * Notify the listeners with given database event, after the customized implementation of AbstractCatalog create/alter/drop a database,
     * it can create the specific event and call the notify method.
     */
    protected void notify(BaseDatabaseEvent event) {
        for (CatalogEventListener listener : listeners) {
            listener.onEvent(event);
        }
    }
}

/* Create default catalog context with listeners. */
public DefaultCatalogContext {
    public DefaultCatalogContext(
            String name,
            Map<String, String> options,
            ReadableConfig configuration,
            ClassLoader classLoader,
            List<CatalogEventListener> listeners);

    /* Get catalog event listeners from the context. */
    public List<CatalogEventListener> listeners() {
        return listeners;
    }
}

/* Set pipeline to client before it submit job graph.  */
public class AbstractSessionClusterExecutor {
    public CompletableFuture<JobClient> execute(
            @Nonnull final Pipeline pipeline,
            @Nonnull final Configuration configuration,
            @Nonnull final ClassLoader userCodeClassloader)
            throws Exception {
        ....

        final ClusterClientProvider<ClusterID> clusterClientProvider =
                    clusterDescriptor.retrieve(clusterID);
        ClusterClient<ClusterID> clusterClient = clusterClientProvider.getClusterClient();

        // Set pipeline to the cluster client before submit job graph
        clusterClient.setPipeline(pipeline); 

        return clusterClient
                    .submitJob(jobGraph)
                    ....  
    }
}

/* Create job submission event and notify listeners before it submit job graph. */
public class RestClusterClient {
    private final List<JobSubmissionListener> listeners;
    private Pipeline pipeline;

    @Override
    public CompletableFuture<JobID> submitJob(@Nonnull JobGraph jobGraph) {
        // Create event and notify listeners before the job graph is submitted.
        JobSubmissionEvent event = createEventFromPipeline(pipeline);
        if (event != null) {
            listeners.forEach(listener -> listener.onEvent(event));
        }
        
        ....;
    }
}

/* Set pipeline to client before it submit job graph.  */
public class DefaultExecutionGraph {
    private final List<JobExecutionListener> executionListeners;

    private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
        ....;
        notifyJobStatusHooks(newState, error);
        // notify job execution listeners
        notifyJobExecutionListeners(current, newState, error);
        ....;
    }

    private void notifyJobExecutionListeners(JobStatus current, JobStatus newState, Throwable error) {
        JobStatusEvent event = create job status event;
        executionListeners.forEach(listener -> listener.onEvent(event));
    }
}