code: @Override
  	public void invoke(IN next, Context context) throws Exception {
  		// propagate asynchronous errors
  		checkErroneous();
  
  		byte[] serializedKey = schema.serializeKey(next);
  		byte[] serializedValue = schema.serializeValue(next);
  		String targetTopic = schema.getTargetTopic(next);
  		if (targetTopic == null) {
  			targetTopic = defaultTopicId;
  		}
  
  		int[] partitions = this.topicPartitionsMap.get(targetTopic);
  		if (null == partitions) {
  			partitions = getPartitionsByTopic(targetTopic, producer);
  			this.topicPartitionsMap.put(targetTopic, partitions);
  		}
  
  		ProducerRecord<byte[], byte[]> record;
  		if (flinkKafkaPartitioner == null) {
  			record = new ProducerRecord<>(targetTopic, serializedKey, serializedValue);
  		} else {
  			record = new ProducerRecord<>(
  					targetTopic,
  					flinkKafkaPartitioner.partition(next, serializedKey, serializedValue, targetTopic, partitions),
  					serializedKey,
  					serializedValue);
  		}
  		if (flushOnCheckpoint) {
  			synchronized (pendingRecordsLock) {
  				pendingRecords++;
  			}
  		}
  		producer.send(record, callback);
  	}
docs: Called when new data arrives to the sink, and forwards it to Kafka.
  
  @param next
  The incoming data