bin/start-seatunnel-flink-13-connector-v2.sh --config ./config/test.config --master yarn-per-job

env {
  parallelism = 2
  job.mode = "STREAMING"
#   flink.execution.checkpointing.interval=5000
}
source {
  Kafka {
    schema = {
      fields {
        comment_num = string
        insert_time = string
        user_info = {
            username = string,
            age = string
        }
      }
    }
    topic = "test-topic"
    consumer.group	= "test-group"
    bootstrap.servers = "xxxx"
    kafka.config = {
      client.id = client_1
      max.poll.records = 500
      auto.offset.reset = "earliest"
      enable.auto.commit = "false"
    }
    result_table_name = "kafka_table"
  }
}



sink {
    Elasticsearch {
        source_table_name = "kafka_table"
        hosts = ["xxxx"]
        index = "test-588"
    }
}

    @Override
    public Optional<ElasticsearchCommitInfo> prepareCommit() {
        bulkEsWithRetry(this.esRestClient, this.requestEsList);
        return Optional.empty();
    }

    @Override
    public void write(SeaTunnelRow element) {
        if (RowKind.UPDATE_BEFORE.equals(element.getRowKind())) {
            return;
        }

        String indexRequestRow = seaTunnelRowSerializer.serializeRow(element);
        requestEsList.add(indexRequestRow);
        if (requestEsList.size() >= maxBatchSize) {
            bulkEsWithRetry(this.esRestClient, this.requestEsList);
        }
    }

seatunnel.yaml

seatunnel:
  engine:
    history-job-expire-minutes: 1440
    backup-count: 1
    queue-type: blockingqueue
    print-execution-info-interval: 60
    print-job-metrics-info-interval: 60
    slot-service:
      dynamic-slot: true
    checkpoint:
      interval: 10000
      timeout: 60000
      storage:
        type: hdfs
        max-retained: 3
        plugin-config:
          namespace: /tmp/seatunnel/checkpoint_snapshot
          storage.type: hdfs

### Running Command

```shell
bin/start-seatunnel-flink-13-connector-v2.sh --config ./config/test.config --master yarn-per-job

When the data does not meet the maxBatchSize, the writer will not write the data. This will cause the data to not be flushed.