### single fat jar file
PIPELINE_FAT_JAR = "pyflink-getting-started-1.jar"
table_env.get_config().set(
    "pipeline.jars", f"file://{os.path.join(CURRENT_DIR, 'lib', PIPELINE_FAT_JAR)}"
)

### multiple JAR files
CURRENT_DIR = os.path.dirname(os.path.realpath(__file__))
FLINK_SQL_CONNECTOR_KAFKA = "flink-sql-connector-kafka-1.15.2.jar"
AWS_MSK_IAM_AUTH = "aws-msk-iam-auth-1.1.7-all.jar"
KAFKA_CLIENTS = "kafka-clients-2.8.1.jar"
table_env.get_config().set(
    "pipeline.jars",
    f"file://{os.path.join(CURRENT_DIR, 'lib', FLINK_SQL_CONNECTOR_KAFKA)};file://{os.path.join(CURRENT_DIR, 'lib', AWS_MSK_IAM_AUTH)};file://{os.path.join(CURRENT_DIR, 'lib', KAFKA_CLIENTS)}",
)

### table definition
def create_source_table(
    table_name: str, topic_name: str, bootstrap_servers: str, startup_mode: str
):
    return f"""
    CREATE TABLE {table_name} (
        event_time TIMESTAMP(3),
        ticker VARCHAR(6),
        price DOUBLE
    )
    WITH (
        'connector' = 'kafka',
        'topic' = '{topic_name}',
        'properties.bootstrap.servers' = '{bootstrap_servers}',
        'properties.group.id' = 'source-group',
        'format' = 'json',
        'scan.startup.mode' = '{startup_mode}',
        'properties.security.protocol' = 'SASL_SSL',
        'properties.sasl.mechanism' = 'AWS_MSK_IAM',
        'properties.sasl.jaas.config' = 'software.amazon.msk.auth.iam.IAMLoginModule required;',
        'properties.sasl.client.callback.handler.class' = 'software.amazon.msk.auth.iam.IAMClientCallbackHandler'
    )
    """