docs: @Override
  	public void run() {
  		LOG.info("Starting to fetch from {}", this.partitions);
  
  		// set up the config values
  		try {
  			// create the Kafka consumer that we actually use for fetching
  			consumer = new SimpleConsumer(broker.host(), broker.port(), soTimeout, bufferSize, clientId);
  
  			// replace earliest of latest starting offsets with actual offset values fetched from Kafka
  			requestAndSetEarliestOrLatestOffsetsFromKafka(consumer, partitions);
  
  			LOG.info("Starting to consume {} partitions with consumer thread {}", partitions.size(), getName());
  
  			// Now, the actual work starts :-)
  			int offsetOutOfRangeCount = 0;
  			int reconnects = 0;
  			while (running) {
  
  				// ----------------------------------- partitions list maintenance ----------------------------
  
  				// check queue for new partitions to read from:
  				List<KafkaTopicPartitionState<TopicAndPartition>> newPartitions = newPartitionsQueue.pollBatch();
  				if (newPartitions != null) {
  					// found some new partitions for this thread's broker
  
  					// the new partitions should already be assigned a starting offset
  					checkAllPartitionsHaveDefinedStartingOffsets(newPartitions);
  					// if the new partitions are to start from earliest or latest offsets,
  					// we need to replace them with actual values from Kafka
  					requestAndSetEarliestOrLatestOffsetsFromKafka(consumer, newPartitions);
  
  					// add the new partitions (and check they are not already in there)
  					for (KafkaTopicPartitionState<TopicAndPartition> newPartition: newPartitions) {
  						if (partitions.contains(newPartition)) {
  							throw new IllegalStateException("Adding partition " + newPartition +
  									" to subscribed partitions even though it is already subscribed");
  						}
  						partitions.add(newPartition);
  					}
  
  					LOG.info("Adding {} new partitions to consumer thread {}", newPartitions.size(), getName());
  					LOG.debug("Partitions list: {}", newPartitions);
  				}
  
  				if (partitions.size() == 0) {
  					if (newPartitionsQueue.close()) {
  						// close succeeded. Closing thread
  						running = false;
  
  						LOG.info("Consumer thread {} does not have any partitions assigned anymore. Stopping thread.",
  								getName());
  
  						// add the wake-up marker into the queue to make the main thread
  						// immediately wake up and termination faster
  						unassignedPartitions.add(MARKER);
  
  						break;
  					} else {
  						// close failed: fetcher main thread concurrently added new partitions into the queue.
  						// go to top of loop again and get the new partitions
  						continue;
  					}
  				}
  
  				// ----------------------------------- request / response with kafka ----------------------------
  
  				FetchRequestBuilder frb = new FetchRequestBuilder();
  				frb.clientId(clientId);
  				frb.maxWait(maxWait);
  				frb.minBytes(minBytes);
  
  				for (KafkaTopicPartitionState<?> partition : partitions) {
  					frb.addFetch(
  							partition.getKafkaTopicPartition().getTopic(),
  							partition.getKafkaTopicPartition().getPartition(),
  							partition.getOffset() + 1, // request the next record
  							fetchSize);
  				}
  
  				kafka.api.FetchRequest fetchRequest = frb.build();
  				LOG.debug("Issuing fetch request {}", fetchRequest);
  
  				FetchResponse fetchResponse;
  				try {
  					fetchResponse = consumer.fetch(fetchRequest);
  				}
  				catch (Throwable cce) {
  					//noinspection ConstantConditions
  					if (cce instanceof ClosedChannelException) {
  						LOG.warn("Fetch failed because of ClosedChannelException.");
  						LOG.debug("Full exception", cce);
  
  						// we don't know if the broker is overloaded or unavailable.
  						// retry a few times, then return ALL partitions for new leader lookup
  						if (++reconnects >= reconnectLimit) {
  							LOG.warn("Unable to reach broker after {} retries. Returning all current partitions", reconnectLimit);
  							for (KafkaTopicPartitionState<TopicAndPartition> fp: this.partitions) {
  								unassignedPartitions.add(fp);
  							}
  							this.partitions.clear();
  							continue; // jump to top of loop: will close thread or subscribe to new partitions
  						}
  						try {
  							consumer.close();
  						} catch (Throwable t) {
  							LOG.warn("Error while closing consumer connection", t);
  						}
  						// delay & retry
  						Thread.sleep(100);
  						consumer = new SimpleConsumer(broker.host(), broker.port(), soTimeout, bufferSize, clientId);
  						continue; // retry
  					} else {
  						throw cce;
  					}
  				}
  				reconnects = 0;
  
  				// ---------------------------------------- error handling ----------------------------
  
  				if (fetchResponse == null) {
  					throw new IOException("Fetch from Kafka failed (request returned null)");
  				}
  
  				if (fetchResponse.hasError()) {
  					String exception = "";
  					List<KafkaTopicPartitionState<TopicAndPartition>> partitionsToGetOffsetsFor = new ArrayList<>();
  
  					// iterate over partitions to get individual error codes
  					Iterator<KafkaTopicPartitionState<TopicAndPartition>> partitionsIterator = partitions.iterator();
  					boolean partitionsRemoved = false;
  
  					while (partitionsIterator.hasNext()) {
  						final KafkaTopicPartitionState<TopicAndPartition> fp = partitionsIterator.next();
  						short code = fetchResponse.errorCode(fp.getTopic(), fp.getPartition());
  
  						if (code == ErrorMapping.OffsetOutOfRangeCode()) {
  							// we were asked to read from an out-of-range-offset (maybe set wrong in Zookeeper)
  							// Kafka's high level consumer is resetting the offset according to 'auto.offset.reset'
  							partitionsToGetOffsetsFor.add(fp);
  						}
  						else if (code == ErrorMapping.NotLeaderForPartitionCode() ||
  								code == ErrorMapping.LeaderNotAvailableCode() ||
  								code == ErrorMapping.BrokerNotAvailableCode() ||
  								code == ErrorMapping.UnknownCode()) {
  							// the broker we are connected to is not the leader for the partition.
  							LOG.warn("{} is not the leader of {}. Reassigning leader for partition", broker, fp);
  							LOG.debug("Error code = {}", code);
  
  							unassignedPartitions.add(fp);
  
  							partitionsIterator.remove(); // unsubscribe the partition ourselves
  							partitionsRemoved = true;
  						}
  						else if (code != ErrorMapping.NoError()) {
  							exception += "\nException for " + fp.getTopic() + ":" + fp.getPartition() + ": " +
  									ExceptionUtils.stringifyException(ErrorMapping.exceptionFor(code));
  						}
  					}
  					if (partitionsToGetOffsetsFor.size() > 0) {
  						// safeguard against an infinite loop.
  						if (offsetOutOfRangeCount++ > 3) {
  							throw new RuntimeException("Found invalid offsets more than three times in partitions "
  									+ partitionsToGetOffsetsFor + " Exceptions: " + exception);
  						}
  						// get valid offsets for these partitions and try again.
  						LOG.warn("The following partitions had an invalid offset: {}", partitionsToGetOffsetsFor);
  						requestAndSetSpecificTimeOffsetsFromKafka(consumer, partitionsToGetOffsetsFor, invalidOffsetBehavior);
  
  						LOG.warn("The new partition offsets are {}", partitionsToGetOffsetsFor);
  						continue; // jump back to create a new fetch request. The offset has not been touched.
  					}
  					else if (partitionsRemoved) {
  						continue; // create new fetch request
  					}
  					else {
  						// partitions failed on an error
  						throw new IOException("Error while fetching from broker '" + broker + "': " + exception);
  					}
  				} else {
  					// successful fetch, reset offsetOutOfRangeCount.
  					offsetOutOfRangeCount = 0;
  				}
  
  				// ----------------------------------- process fetch response ----------------------------
  
  				int messagesInFetch = 0;
  				int deletedMessages = 0;
  				Iterator<KafkaTopicPartitionState<TopicAndPartition>> partitionsIterator = partitions.iterator();
  
  				partitionsLoop:
  				while (partitionsIterator.hasNext()) {
  					final KafkaTopicPartitionState<TopicAndPartition> currentPartition = partitionsIterator.next();
  
  					final ByteBufferMessageSet messageSet = fetchResponse.messageSet(
  							currentPartition.getTopic(), currentPartition.getPartition());
  
  					for (MessageAndOffset msg : messageSet) {
  						if (running) {
  							messagesInFetch++;
  							final ByteBuffer payload = msg.message().payload();
  							final long offset = msg.offset();
  
  							if (offset <= currentPartition.getOffset()) {
  								// we have seen this message already
  								LOG.info("Skipping message with offset " + msg.offset()
  										+ " because we have seen messages until (including) "
  										+ currentPartition.getOffset()
  										+ " from topic/partition " + currentPartition.getTopic() + '/'
  										+ currentPartition.getPartition() + " already");
  								continue;
  							}
  
  							// If the message value is null, this represents a delete command for the message key.
  							// Log this and pass it on to the client who might want to also receive delete messages.
  							byte[] valueBytes;
  							if (payload == null) {
  								deletedMessages++;
  								valueBytes = null;
  							} else {
  								valueBytes = new byte[payload.remaining()];
  								payload.get(valueBytes);
  							}
  
  							// put key into byte array
  							byte[] keyBytes = null;
  							int keySize = msg.message().keySize();
  
  							if (keySize >= 0) { // message().hasKey() is doing the same. We save one int deserialization
  								ByteBuffer keyPayload = msg.message().key();
  								keyBytes = new byte[keySize];
  								keyPayload.get(keyBytes);
  							}
  
  							final T value = deserializer.deserialize(
  								new ConsumerRecord<>(
  									currentPartition.getTopic(),
  									currentPartition.getPartition(), keyBytes, valueBytes, offset));
  
  							if (deserializer.isEndOfStream(value)) {
  								// remove partition from subscribed partitions.
  								partitionsIterator.remove();
  								continue partitionsLoop;
  							}
  
  							owner.emitRecord(value, currentPartition, offset);
  						}
  						else {
  							// no longer running
  							return;
  						}
  					}
  				}
  				LOG.debug("This fetch contained {} messages ({} deleted messages)", messagesInFetch, deletedMessages);
  			} // end of fetch loop
  
  			if (!newPartitionsQueue.close()) {
  				throw new Exception("Bug: Cleanly leaving fetcher thread without having a closed queue.");
  			}
  		}
  		catch (Throwable t) {
  			// report to the fetcher's error handler
  			errorHandler.reportError(t);
  		}
  		finally {
  			if (consumer != null) {
  				// closing the consumer should not fail the program
  				try {
  					consumer.close();
  				}
  				catch (Throwable t) {
  					LOG.error("Error while closing the Kafka simple consumer", t);
  				}
  			}
  		}
  	}