/**
 * When job is created or its status is changed, Flink will generate job event and notify job status changed listener.
 */
@PublicEvolving
public interface JobStatusChangedListener {
    /* Event will be fired when job status is changed. */
    public void onEvent(JobStatusChangedEvent event);
}

/** Basic job status event. */
@PublicEvolving
public abstract class JobStatusChangedEvent {
    JobID jobId();
    String jobName();
}

/** Job created event with job lineage. */
@PublicEvolving
public class JobCreatedEvent extends JobStatusChangedEvent {
    /* Lineage for the current job. */
    JobLineage lineage()

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

    /* CreateJob catalogtype, managerTableOrSQL withor listener listDataStream. */
    privateString CatalogManagerjobType();

    /* Job execution type, BATCH or STREAMING. */
 String defaultCatalogName,
  String executionType(); 
     
   Catalog defaultCatalog,
/* Job configuration. */
    Map<String, String> config();   DataTypeFactory typeFactory,
            ManagedTableListener managedTableListener,
}

/** Job status changed event for runtime. */
@PublicEvolving
public class JobExecutionStatusEvent extends JobStatusChangedEvent {
    JobStatus oldStatus();
    JobStatus newStatus();
    List<CatalogModificationListener> listeners@Nullable Throwable exception();
}

    /** NotifyFactory thefor listenersjob withstatus givenchanged catalog eventlistener. */
@PublicEvolving
public    private void notify(CatalogEvent event)interface JobStatusChangedListenerFactory {
    JobStatusChangedListener createListener(Configuration config,    listeners.forEach(listener -> listener.onEvent(event));
    }

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

/**
 * Job lineage is built according to StreamGraph. Users can get sources, sinks and relationships from lineage.
 */
@PublicEvolvig
public class JobLineage {
    /* Source lineage entity list. */
    List<LineageEntity> sources();

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

    /* AddLineage databaserelations ddlsfrom andsources notify listener for databasesto sinks. */
    public void createDatabase/dropDatabase/alterDatabase(...) {
        ....;
        notify(Create Different Database Modification Event); 
    }

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

/** Basic table lineage entity which has catalog context and table in it. */
public class TableLineageEntity extends LineageEntity {
    /* The catalog context of the table lineage entity. */
    public CatalogContext catalogContext();

    /* The table of the table lineage entity. */
    public CatalogBaseTable table();
}

/** Source lineage entity for table. */
@PublicEvolving
public class TableSourceLineageEntity extends TableLineageEntity {
}

/** Sink lineage entity for table. */
@PublicEvolving
public class TableSinkLineageEntity extends TableLineageEntity {
    /* Modify type, INSERT/UPDATE/DELETE. */
    String modifyType();
 
    /* Update mode, APPEND/RETRACT/UPSERT. */
    String updateMode();
    boolean overwrite();
}

/* Table lineage relations from source table to sink table. */
@PublicEvolving
public class TableLineageRelation extends LineageRelation {
    /* Table column lineage relations from source to sink. */
    List<TableColumnLineageRelation> columnRelations();
}

/* Column lineage from source table to sink table. */
@PublicEvolving
public class TableColumnLineageRelation {
    /* Source table column. */
    String sourceColumn();

    /* Sink table column. */
    String sinkColumn();
}

/**
 * Add setLineageEntity method in data stream source.
 */
@Public
public class DataStreamSource {
    private LineageEntity lineageEntity;
 
    public DataStreamSource setLineageEntity(LineageEntity lineageEntity);
}
 
/**
 * Add setLineageEntity and addLineageRelations methods in data stream sink.
 */
@Public
public class DataStreamSink {
    private LineageEntity lineageEntity;
 
    public DataStreamSink setLineageEntity(LineageEntity lineageEntity);

    /* Add lineage relations for data stream jobs. */
    public DataStreamSink addLineageRelations(LineageRelation ... relations);
}

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

# Config for job status changed listeners.
lineage.job-status-changed.listeners: {job status changed listener factory1},{job status changed listener factory2}

/** User defined vector source and sink lineage entity. */
public class KafkaVectorLineageEntity extends LineageEntity {
    /* The capacity of source lineage. */
    int capacity();

    /* The value type in the vector. */
    String valueType();
}
 
/* User can use vector source/sink lineages in datastream job. */
Map<String, String> kafkaSourceConf = ...;
Map<String, String> kafkaSinkConf = ...;
StreamExecutionEnvironment env = ...;
KafkaVectorLineageEntity sourceLineageEntity = new KafkaVectorLineageEntity(10, "int", kafkaSourceConf);
KafkaVectorLineageEntity sinkLineageEntity = new KafkaVectorLineageEntity(20, "double", kafkaSinkConf);
env.fromSource(...)
        .setLineageEntity(sourceLineageEntity) // Set source lineage entity
	.map(...).keyBy(..).reduce(..)...
	.sinkTo(...)
        .setLineageEntity(sinkLineageEntity)  // Set sink lineage entity
        .addLineageRelations(                 // Add lineage relations from sources to the current sink
            new LineageRelation(sourceLineageEntity, sinkLineageEntity);
env.execute();

-- 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(CatalogEvent event) {
        listeners.forEach(listener -> listener.onEvent(event));
    }

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

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

    /* Add listeners in Builder for catalog manager. */
    public static final class Builder {
        Builder listeners(List<CatalogModificationListener> 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<JobStatusChangedListener> 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<JobStatusChangedListener> executionListeners;

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

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