def collect(self) -> 'CloseableIterator':
    """
    Get the result contents as a closeable row iterator.

    Note:

    For SELECT operation, the job will not be finished unless all result data has been
    collected. So we should actively close the job to avoid resource leak through
    CloseableIterator#close method. Calling CloseableIterator#close method will cancel the job
    and release related resources.

    For DML operation, Flink does not support getting the real affected row count now. So the
    affected row count is always -1 (unknown) for every sink, and them will be returned until
    the job is finished.
    Calling CloseableIterator#close method will cancel the job.

    For other operations, no flink job will be submitted (get_job_client() is always empty), and
    the result is bounded. Do noting when calling CloseableIterator#close method.

    Recommended code to call CloseableIterator#close method looks like:

    >>> table_result = t_env.execute("select ...")
    >>> with table_result.collect() as results:
    >>>    for result in results:
    >>>        ...

    In order to fetch result to local, you can call either collect() and print(). But, they can
    not be called both on the same TableResult instance.

    :return: A CloseableIterator.

    .. versionadded:: 1.12.0
    """

version: "2.2"
services:
  jobmanager:
    build: .
    image: varimatflink:0.1
    hostname: jobmanager
    container_name: jobmanager
    ports:
      - "8081:8081"
    command: jobmanager
    volumes:
      - ./data:/data
      - ./artifacts:/opt/flink/usrlib
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        parallelism.default: 2        

  taskmanager:
    build: .
    image: varimatflink:0.1
    hostname: taskmanager
    container_name: taskmanager
    depends_on:
      - jobmanager
    command: taskmanager
    volumes:
      - ./data:/data
      - ./artifacts:/opt/flink/usrlib
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 4
        parallelism.default: 2

  zookeeper:
    image: confluentinc/cp-zookeeper:6.1.1
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    restart: always

  broker:
    image: confluentinc/cp-kafka:6.1.1
    hostname: kafka
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: <PLAINTEXT://kafka:9092>,PLAINTEXT_<HOST://localhost:29092>
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0