code: def test_kafka_batch_checkpointing_async_nodes_2():
      """
      In async nodes like zip_latest, zip, combine_latest which involve multiple streams,
      checkpointing in each stream commits offsets after the datum in that specific stream
      is processed completely and exits the pipeline.
      """
      CONSUMER_ARGS1 = {
          "bootstrap.servers": "localhost:9092",
          "group.id": "zip_latest",
          "enable.auto.commit": False,
      }
      CONSUMER_ARGS2 = {
          "bootstrap.servers": "localhost:9092",
          "group.id": "zip",
          "enable.auto.commit": False,
      }
      CONSUMER_ARGS3 = {
          "bootstrap.servers": "localhost:9092",
          "group.id": "combine_latest",
          "enable.auto.commit": False,
      }
      TOPIC1 = "test1"
      TOPIC2 = "test2"
      with kafka_service() as kafka:
          kafka, TOPIC = kafka
          stream1 = Stream.from_kafka_batched(
              TOPIC1, CONSUMER_ARGS1, asynchronous=True
          )
          stream1.start()
          stream2 = Stream.from_kafka_batched(
              TOPIC2, CONSUMER_ARGS1, asynchronous=True
          )
          stream2.start()
          stream1.zip_latest(stream2).sink_to_list()
          stream3 = Stream.from_kafka_batched(
              TOPIC1, CONSUMER_ARGS2, asynchronous=True
          )
          stream3.start()
          stream4 = Stream.from_kafka_batched(
              TOPIC2, CONSUMER_ARGS2, asynchronous=True
          )
          stream4.start()
          stream3.zip(stream4).sink_to_list()
          stream5 = Stream.from_kafka_batched(
              TOPIC1, CONSUMER_ARGS3, asynchronous=True
          )
          stream5.start()
          stream6 = Stream.from_kafka_batched(
              TOPIC2, CONSUMER_ARGS3, asynchronous=True
          )
          stream6.start()
          stream5.combine_latest(stream6).sink_to_list()
          kafka.produce(TOPIC1, b"value-0")
          time.sleep(5)
          kafka.produce(TOPIC2, b"value-1")
          """
          1. zip_latest emits a tuple, the lossless stream 1 commits an offset.
          2. Since zip emits a tuple, streams 3 and 4 commit offsets in their topics.
          3. combine_latest does not commit any offset since data is still to be used.
          """
          time.sleep(5)
          kafka.produce(TOPIC1, b"value-2")
          """
          1. zip_latest emits a tuple, the lossless stream 1 commits an offset.
          2. zip does not commit any offset.
          3. combine_latest commits an offset in stream 5.
          """
          time.sleep(5)
          kafka.produce(TOPIC2, b"value-3")
          """
          1. zip_latest emits a tuple, the non-lossless stream 2 commits an offset.
          2. Since zip emits a tuple, streams 3 and 4 commit offsets in their topics.
          3. combine_latest commits an offset in stream 6.
          """
          time.sleep(10)
          stream1.upstream.stopped = True
          stream2.upstream.stopped = True
          stream3.upstream.stopped = True
          stream4.upstream.stopped = True
          stream5.upstream.stopped = True
          stream6.upstream.stopped = True
          stream1.destroy()
          stream2.destroy()
          stream3.destroy()
          stream4.destroy()
          stream5.destroy()
          stream6.destroy()
          """
          Each stream/group.id picks up from their last committed offset.
          """
          consumer1 = ck.Consumer(CONSUMER_ARGS1)
          consumer2 = ck.Consumer(CONSUMER_ARGS2)
          consumer3 = ck.Consumer(CONSUMER_ARGS3)
          tps = [
              ck.TopicPartition(TOPIC1, 0),
              ck.TopicPartition(TOPIC2, 0),
          ]
          committed1 = consumer1.committed(tps)
          committed2 = consumer2.committed(tps)
          committed3 = consumer3.committed(tps)
          assert committed1[0].offset == 2
          assert committed1[1].offset == 1
          assert committed2[0].offset == 2
          assert committed2[1].offset == 2
          assert committed3[0].offset == 1
          assert committed3[1].offset == 1
  
imports: import asyncio
  import atexit
  from contextlib import contextmanager
  from flaky import flaky
  import os
  import pytest
  import random
  import shlex
  import subprocess
  import time
  from tornado import gen
  from ..core import Stream
  from ..dask import DaskStream
  from streamz.utils_test import gen_test, wait_for, await_for
  from distributed.utils_test import gen_cluster
label_desc: Use image node_id 13 for calling a global function with example usage: test_kafka_batch_checkpointing_async_nodes_2() without return types