code: def test_kafka_batch_checkpointing_async_nodes_1():
      """
      In async nodes like partition & sliding window, data is checkpointed only after
      the pipeline has finished processing it.
      """
      j = random.randint(0, 10000)
      ARGS = {
          "bootstrap.servers": "localhost:9092",
          "group.id": "streamz-test%i" % j,
          "enable.auto.commit": False,
      }
      with kafka_service() as kafka:
          kafka, TOPIC = kafka
          stream1 = Stream.from_kafka_batched(TOPIC, ARGS)
          out1 = (
              stream1.partition(2)
              .sliding_window(2, return_partial=False)
              .sink_to_list()
          )
          stream1.start()
          for i in range(0, 2):
              kafka.produce(TOPIC, b"value-%d" % i)
              kafka.flush()
              time.sleep(2)
          assert len(out1) == 0
          # Stream stops before data can finish processing, hence no checkpointing.
          stream1.upstream.stopped = True
          stream1.destroy()
          stream2 = Stream.from_kafka_batched(TOPIC, ARGS)
          out2 = (
              stream2.partition(2)
              .sliding_window(2, return_partial=False)
              .sink_to_list()
          )
          stream2.start()
          wait_for(lambda: stream2.upstream.started, 10, 0.1)
          for i in range(2, 6):
              kafka.produce(TOPIC, b"value-%d" % i)
              kafka.flush()
              time.sleep(2)
          assert len(out2) == 1
          assert out2 == [
              (
                  ([b"value-0", b"value-1"], [b"value-2"]),
                  ([b"value-3"], [b"value-4"]),
              )
          ]
          # Some data gets processed and exits pipeline before the stream stops, hence checkpointing complete.
          stream2.upstream.stopped = True
          stream2.destroy()
          stream3 = Stream.from_kafka_batched(TOPIC, ARGS)
          out3 = stream3.sink_to_list()
          stream3.start()
          wait_for(lambda: stream3.upstream.started, 10, 0.1)
          # Stream picks up from where it left before, i.e., from the last committed offset.
          wait_for(
              lambda: len(out3) == 1
              and out3[0] == [b"value-3", b"value-4", b"value-5"],
              10,
              0.1,
          )
          stream3.upstream.stopped = True
          stream3.destroy()
  
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 12 for calling a global function with example usage: test_kafka_batch_checkpointing_async_nodes_1() without return types