code: private SimpleConsumerThread<T> createAndStartSimpleConsumerThread(
  			List<KafkaTopicPartitionState<TopicAndPartition>> seedPartitions,
  			Node leader,
  			ExceptionProxy errorHandler) throws IOException, ClassNotFoundException {
  		// each thread needs its own copy of the deserializer, because the deserializer is
  		// not necessarily thread safe
  		final KafkaDeserializationSchema<T> clonedDeserializer =
  				InstantiationUtil.clone(deserializer, runtimeContext.getUserCodeClassLoader());
  
  		// seed thread with list of fetch partitions (otherwise it would shut down immediately again
  		SimpleConsumerThread<T> brokerThread = new SimpleConsumerThread<>(
  				this, errorHandler, kafkaConfig, leader, seedPartitions, unassignedPartitionsQueue,
  				clonedDeserializer, invalidOffsetBehavior);
  
  		brokerThread.setName(String.format("SimpleConsumer - %s - broker-%s (%s:%d)",
  				runtimeContext.getTaskName(), leader.id(), leader.host(), leader.port()));
  		brokerThread.setDaemon(true);
  		brokerThread.start();
  
  		LOG.info("Starting thread {}", brokerThread.getName());
  		return brokerThread;
  	}
docs: ------------------------------------------------------------------------