code: def check_cluster(
      cluster_config,
      data_path,
      java_home,
      check_replicas,
      batch_size,
      minutes,
      start_time,
      end_time,
  ):
      
      brokers = get_broker_list(cluster_config)
      broker_files = find_files(data_path, brokers, minutes, start_time, end_time)
      if not check_replicas:  # remove replicas
          broker_files = filter_leader_files(cluster_config, broker_files)
      processes = []
      print("Starting {n} parallel processes".format(n=len(broker_files)))
      try:
          for broker, host, files in broker_files:
              print(
                  "  Broker: {host}, {n} files to check".format(
                      host=host,
                      n=len(files)),
              )
              p = Process(
                  name="dump_process_" + host,
                  target=check_files_on_host,
                  args=(java_home, host, files, batch_size),
              )
              p.start()
              processes.append(p)
          print("Processes running:")
          for process in processes:
              process.join()
      except KeyboardInterrupt:
          print("Terminating all processes")
          for process in processes:
              process.terminate()
              process.join()
          print("All processes terminated")
          sys.exit(1)
docs: Check the integrity of the Kafka log files in a cluster.
  
      start_time and end_time should be in the format specified
      by TIME_FORMAT_REGEX.
  
      :param data_path: the path to the log folder on the broker
      :type data_path: str
      :param java_home: the JAVA_HOME of the broker
      :type java_home: str
      :param check_replicas: also checks the replica files
      :type check_replicas: bool
      :param batch_size: the size of the batch
      :type batch_size: int
      :param minutes: check the files modified in the last N minutes
      :type minutes: int
      :param start_time: check the files modified after start_time
      :type start_time: str
      :param end_time: check the files modified before end_time
      :type end_time: str