code: def test_kafka_checkpointing_auto_offset_reset_latest():
      """
      Testing whether checkpointing works as expected with multiple topic partitions and
      with auto.offset.reset configuration set to latest (also default).
      """
      j = random.randint(0, 10000)
      ARGS = {
          "bootstrap.servers": "localhost:9092",
          "group.id": "streamz-test%i" % j,
          "auto.offset.reset": "latest",
      }
      with kafka_service() as kafka:
          kafka, TOPIC = kafka
          TOPIC = "test-checkpointing-offset-reset-latest"
          subprocess.call(
              shlex.split(
                  "docker exec streamz-kafka "
                  "/opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh "
                  "--create --zookeeper localhost:2181 "
                  "--replication-factor 1 --partitions 3 "
                  "--topic test-checkpointing-offset-reset-latest"
              )
          )
  
          """
          Since the stream has not started yet, these messages are not read because
          the stream has auto.offset.reset set to latest.
          """
          for i in range(30):
              kafka.produce(TOPIC, b"value-%d" % i)
          kafka.flush()
  
          stream1 = Stream.from_kafka_batched(
              TOPIC, ARGS, asynchronous=True
          )
          out1 = stream1.map(split).gather().sink_to_list()
          time.sleep(1)  # messages make ttheir way through kafka
  
          stream1.start()
          wait_for(lambda: stream1.upstream.started, 10, period=0.1)
  
          """
          Stream has started, so these are read.
          """
          for i in range(30):
              kafka.produce(TOPIC, b"value-%d" % i)
          kafka.flush()
  
          wait_for(
              lambda: len(out1) == 3
              and (len(out1[0]) + len(out1[1]) + len(out1[2])) == 30,
              10,
              period=0.1,
          )
          """
          Stream stops but checkpoint has been created.
          """
          stream1.upstream.stopped = True
  
          """
          When the stream is restarted, these messages are read, because the checkpoint
          overrrides the auto.offset.reset:latest config this time around as expected.
          """
          for i in range(30):
              kafka.produce(TOPIC, b"value-%d" % i)
          kafka.flush()
  
          stream2 = Stream.from_kafka_batched(
              TOPIC, ARGS, asynchronous=True
          )
          out2 = stream2.map(split).gather().sink_to_list()
  
          """
          Stream restarts here.
          """
          stream2.start()
          wait_for(lambda: stream2.upstream.started, 10, 0.1)
  
          for i in range(30):
              kafka.produce(TOPIC, b"value-%d" % i)
          kafka.flush()
  
          wait_for(
              lambda: len(out2) == 6
              and (len(out2[3]) + len(out2[4]) + len(out2[5])) == 30,
              10,
              period=0.1,
          )
          stream2.upstream.stopped = True
  
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 14 for calling a global function with example usage: test_kafka_checkpointing_auto_offset_reset_latest() without return types