/**
 * 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

/**
 * Job creation listener, client will create specific event and notify this listener.
 */
@PublicEvolving
public interface JobDeploymentListenerCatalogModificationListener {
	/* Start the listener. */
	void start(Map<String, String> config);

    /* Event fired after a catalog has been registered/** The event will be fired when the database/table is modified. */
    void onRegisterCatalogonEvent(CatalogEventCatalogModificationEvent catalogEventevent);
}

/* Basic interface for void onUnregisterCatalog(UnregisterCatalogEvent catalogEvent);

    /* Event fired after a database has been createdcatalog modification. */
@PublicEvolving
public interface class CatalogModificationEvent {
    /* Context for the event. */
    CatalogContext void onCreateDatabase(DatabaseEvent databaseEventcontext();
}

    /* EventContext firedfor aftercatalog amodification databaseand hasjob beenlineage droppedevents. */
@PublicEvolving
public interface CatalogContext {
 void onDropDatabase(DatabaseEvent databaseEvent);
  
    /* EventThe firedname after a table has been createdof catalog. */
    voidString onCreateTablegetCatalogName(CreateTableEvent tableEvent);

    /* EventClass firedof after a table has been changedthe catalog. */
    void onAlterTable(AlterTableEvent tableEventClass<? extends Catalog> getClass();

    /* EventIdentifier fired after a table has been droppedfor the catalog from catalog factory, such as jdbc/iceberg/paimon. */
    voidOptional<String> onDropTablegetFactoryIdentifier(DropTableEvent tableEvent); 

    /* EventConfig fired before a job is submitted to do some validationsfor catalog. */
    Configuration void onJobPreSubmission(JobSubmissionEvent submitEventgetConfiguration(); 
}

    /* EventThe firedbasic afterclass afor jobdatabase isrelated submittedevent. */
public interface DatabaseModificationEvent extends void onJobSubmission(JobSubmissionEvent submitEvent); CatalogModificationEvent {
 
   CatalogDatabase database();
}

/* Event for catalogcreating registration, provides catalog name, default database, database list and properties in the Catalogdatabase. */
@PublicEvolving
public interface CreateDatabaseEvent extends DatabaseModificationEvent {
    boolean ignoreIfExists();
}

/* Event for altering database. */
    @PublicEvolving
public interface AlterDatabaseEvent extends interface CatalogEventDatabaseModificationEvent {
        String catalogCatalogDatabase newDatabase();
        Catalog catalogboolean ignoreIfNotExists();
    }
 
    /* Event for catalogdropping unregistrationdatabase. */ 
    @PublicEvolving
public interface DropDatabaseEvent extends interface UnregisterCatalogEventDatabaseModificationEvent {
        String catalog();
        boolean ignoreIfNotExistsignoreIfExists();
    }

    /**
 Event* forBase databasetable creationevent, provides catalog namecolumn list, primary keys, databasepartition namekeys, commentwatermarks and properties ofin
 the* databaseCatalogBaseTable. */
    @PublicEvolving
    interface DatabaseEvent The table can be source or sink.
 */
public interface TableModificationEvent extends CatalogModificationEvent {
        String catalogObjectIdentifier identifier();
        String nameCatalogBaseTable table();
        CatalogDatabase database();
    }

/* Event for creating table. */
@PublicEvolving
public interface CreateTableEvent extends CatalogModificationEvent {
    boolean ignoreIfExists();
    }

/* Event for altering table, provides /*all Eventchanges for droppingold databasetable. */
    @PublicEvolving
public interface AlterTableEvent extends interfaceCatalogModificationEvent DropDatabaseEvent {
    List<TableChange> tableChanges();
    Stringboolean catalogignoreIfExists();
}

/* Event for dropping table. */
@PublicEvolving
public interface DropTableEvent String name(); 
    extends CatalogModificationEvent {
    boolean ignoreIfExists();
    }

    /* TableFactory informationfor event,catalog providesmodification column list, primary keys, partition keys, watermarks and properties in the table. The table can be source or sink. */listener. */
@PublicEvolving
public interface CatalogModificationListenerFactory {
    CatalogModificationListener createListener(Context context);

    @PublicEvolving 
    public interface Context {
    @PublicEvolving
    interface TableEvent {Configuration getConfiguration();
        ObjectIdentifier identifier();  
        CatalogBaseTable tableClassLoader getUserClassLoader();
    }
	/*
    /	* EventGet foran tableExecutor creation. */
    @PublicEvolving
    interface CreateTableEvent extends TableEvent {pool for the listener to run async operations that can potentially be IO-heavy.
    	*/
    boolean	Executor ignoreIfExistsgetIOExecutor();
      }

    /* Event for altering table, provides all information in old table and new table. */
    @PublicEvolving
    interface AlterTableEvent extends TableEvent {
        CreateTableEvent newTable();
        boolean ignoreIfExists();
    }

    /* Event for dropping table. */
    @PublicEvolving
    interface DropTableEvent {
        ObjectIdentifier identifier();
        boolean ignoreIfExists();   
    }

    /* Event for before and after job is submitted, provides source tables and sink tables. */
    @PublicEvolving
    interface JobSubmissionEvent {
        JobID jobId();
        String jobName();
        List<TableEvent> sources();
        List<TableEvent> sinks();
    }
}

/* Kafka storage identifier options. */
"properties.bootstrap.servers" for Kafka bootstrap servers
"topic" for Kafka Topic
"properties.group.id" for Kafka group id
"topic-pattern" for Kafka topic pattern

# Config for catalog modification listeners.
table.catalog-modification.listeners: {table catalog listener factory1},{table catalog listener factory2}

-- Create a table my_kafka_table1 for kafka topic 'kafka_topic'
CREATE TABLE my_kafka_table1 (
  val1 STRING,
  val2 STRING,
  val3 STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafka_topic',
  'properties.bootstrap.servers' = 'kafka_bootstrap_server',
  'properties.group.id' = 'kafka_group',
  'format' = 'json'
);

-- Create a Paimon catalog and table for warehouse 'paimon_path1'
CREATE CATALOG paimon_catalog WITH (
    'type'='paimon',
    'warehouse'='paimon_path1'
);
USE CATALOG paimon_catalog;
CREATE TABLE my_paimon_table (
    val1 STRING,
    val2 STRING,
    val3 STRING
) WITH (...);

-- Insert data to Paimon table from Kafka
INSERT INTO my_paimon_table SELECT ... FROM default_catalog.default_database.my_kafka_table1 WHERE ...;

-- Create another table my_kafka_table2 for kafka topic 'kafka_topic' which is same as my_kafka_table1 above
CREATE TABLE my_kafka_table2 (
  val1 STRING,
  val2 STRING,
  val3 STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafka_topic',
  'properties.bootstrap.servers' = 'kafka_bootstrap_server',
  'properties.group.id' = 'kafka_group',
  'format' = 'json'
);

-- Create a Paimon catalog with the same name 'paimon_catalog' for different warehouse 'paimon_path2'
CREATE CATALOG paimon_catalog WITH (
    'type'='paimon',
    'warehouse'='paimon_path2'
);
USE CATALOG paimon_catalog;
CREATE TABLE my_paimon_table (
    val1 STRING,
    val2 STRING,
    val3 STRING
) WITH (...);

-- Insert data to Paimon table from Kafka
INSERT INTO my_paimon_table SELECT ... FROM default_catalog.default_database.my_kafka_table2 WHERE ...;

-- Create two catalogs for warehouse 'paimon_path1' and 'paimon_path2', there are two different tables 'my_paimon_table'
CREATE CATALOG paimon_catalog1 WITH (
    'type'='paimon',
    'warehouse'='paimon_path1'
);
CREATE CATALOG paimon_catalog2 WITH (
    'type'='paimon',
    'warehouse'='paimon_path2'
);

-- Create mysql table
CREATE TABLE mysql_table (
    val1 STRING,
    val2 STRING,
    val3 STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://HOST:PORT/my_database',
    'table-name' = 'my_mysql_table',
    'username' = '***',
    'password' = '***'
);

-- Insert data to mysql table from two paimon tales
INSERT INTO mysql_table 
    SELECT ... FROM paimon_catalog1.default.my_paimon_table
        JOIN paimon_catalog2.default.my_paimon_table
    ON ... WHERE ...;

/* Listeners and related operations in the catalog manager. */
public final class CatalogManager {
    private final List<CatalogModificationListener> listeners;

    /* Create catalog manager with listener list. */
    private CatalogManager(
            String defaultCatalogName,
            Catalog defaultCatalog,
            DataTypeFactory typeFactory,
            ManagedTableListener managedTableListener,
            List<CatalogModificationListener> listeners);

    /* Notify the listeners with given catalog event. */
    private void notify(CatalogModificationEvent event) {
        listeners.forEach(listener -> listener.onEvent(event));
    }

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

    /* Add database ddls and notify listener for databases. */
    public void createDatabase/dropDatabase/alterDatabase(...) {
        ....;
        notify(Create Different Database Modification Event With Context); 
    }

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