K-ConnectSink-何时启动下游流程？

K-Connect Sink 组件允许将数据传输到外部系统。通常情况下，下游系统需要所有数据都接收完毕后才能执行后续处理。为了确保下游流程不会在数据传输尚未完成时启动，可以在代码中实现一个计数器来跟踪数据传输的状态。

示例代码如下：

from google.cloud import bigquery
from kfp import dsl
from kfp.gcp import use_gcp_secret

PROJECT_ID = 'your_project_id'
DATASET = 'your_dataset_name'
TABLE_NAME = 'your_table_name'
OUTPUT_PATH = 'gs://your_output_path'

@dsl.pipeline(name='my-pipeline')
def my_pipeline():
    # Define a counter to track the status of data transmission
    counter = dsl.ContainerOp(
        name='counter',
        image='busybox:1.28',
        command=['sh', '-c'],
        arguments=['echo "0" > /tmp/count'],
    )

    # Define a query to retrieve data from BigQuery
    bq_query = 'SELECT * FROM {}.{}'.format(DATASET, TABLE_NAME)

    # Define a BigQuery source
    bq_source = dsl.ContainerOp(
        name='bq-source',
        image='gcr.io/cloud-builders/gcloud',
        command=['bash', '-c'],
        arguments=['bq extract --destination_format NEWLINE_DELIMITED_JSON {} {}'.format(bq_query, OUTPUT_PATH)],
        secret=use_gcp_secret('user-gcp-sa'),
    ).after(counter)

    # Define a K-Connect Sink to transfer data to external system
    kconnect_sink = dsl.ContainerOp(
        name='kconnect-sink',
        image='gcr.io/kubeflow-examples/kafka-train:1.0',
        command=['python', 'kconnect_sink.py'],
        arguments=[
            '--kafka_brokers', 'broker1,broker2',
            '--topic', 'my_topic',
            '--input_file', '{}/*'.format(OUTPUT_PATH),
        ],
    ).after(bq_source)

    # Define a logic to update counter
    update_counter = d