private boolean isTransactional = config.getTransactionalProducer();
    private boolean transactionInProgress = false;
	protected Map<TopicPartition, OffsetAndMetadata> offsetsMap = new HashMap<>();
    private Set<TopicPartition> taskTopicPartitions;
	private KafkaProducer<byte[], byte[]> producer;
    private String connectConsumerGroup;

    @Override
	public void start(Map<String, String> props) {
		config = new MirrorTaskConfig(props);
	    taskTopicPartitions = config.taskTopicPartitions();
        isTransactional = config.transactionalProducer();
	    producer = initProducer(isTransactional);
        connectConsumerGroup = getSourceConsumerGroupId();
		if (isTransactional) {
	        loadContextOffsets();
        }
    }

    @Override
    public void open(Collection<TopicPartition> partitions) {
		if (isTransactional) {
    	    loadContextOffsets();
		}
    }

    private void loadContextOffsets() {
		Map<TopicPartition, OffsetAndMetadata> initOffsetsOnTarget = listTargetConsumerGroupOffsets(connectConsumerGroup);
	
        Set<TopicPartition> assignments = context.assignment();

        // only keep the offsets of the partitions assigned to this task
        Map<TopicPartition, Long> contextOffsets = assignments.stream()
        		                                              .filter(x -> currentOffsets.containsKey(x))
                                                              .collect(Collectors.toMap(
        		    private boolean isTransactional = config.getTransactionalProducer();
    private boolean transactionStart = false;
	protected Map<TopicPartition, OffsetAndMetadata> offsetsMap = new HashMap<>();
    private Set<TopicPartition> taskTopicPartitions;
	private KafkaProducer<byte[], byte[]> producer;
    private String consumerGroupId;

    @Override
	public void start(Map<String, String> props) {
		config = new MirrorTaskConfig(props);
	    taskTopicPartitions = config.taskTopicPartitions();
        isTransactional = config.transactionalProducer();
	    producer = initProducer(isTransactional);
        consumerGroupId = getSourceConsumerGroupId();
		if (isTransactional) {
			Map<TopicPartition, OffsetAndMetadata> offsetsOnTarget = listTargetConsumerGroupOffsets(consumerGroupId);
			Map<TopicPartition, Long> offsets = new HashMap<>();
			
			for (Map.Entry<TopicPartition, OffsetAndMetadata> offset : offsetsOnTarget.entrySet()) {
				offsets.put(offset.getKey(), offset.getValue().offset());
			}
			Map<TopicPartition, Long> taskOffsets = loadOffsets(taskTopicPartitions, offsets);
			context.offset(taskOffsets);
               x -> x, x -> currentOffsets.get(x)));

        }context.offset(contextOffsets);
    }

	protected KafkaProducer<byte[], byte[]> initProducer(boolean isTransactional) {
        Map<String, Object> producerConfig = config.targetProducerConfig();
        if (isTransactional) {
        	log.info("use transactional producer"String transactionId = getTransactionId();
        	log.info("use transactional producer with Id:  producerConfig.put(ProducerConfig.ACKS_CONFIG, "all"{} ", transactionId);
            producerConfig.put(ProducerConfig.ENABLEACKS_IDEMPOTENCE_CONFIG, "trueall");
         	   producerConfig.put(ProducerConfig.TRANSACTIONALENABLE_IDIDEMPOTENCE_CONFIG, genTransactionId()"true");
        	producerConfig.put(ProducerConfig.CLIENTTRANSACTIONAL_ID_CONFIG, "trasactional-producer"transactionId);
        } else {
        	producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, "non-trasactional-producer"transactionId);
        }
        return MirrorUtils.newProducer(producerConfig);
	}

    @Override
	public void put(Collection<SinkRecord> records) {
        log.info("receive {} messages from consumer", records.size());/**
     * Per some articles, to avoid ProducerFencedException, transaction id is suggested to set application name + hostname
     * Each MirrorSinkTask if (records.size() == 0) {
        	return;
        }
        try {is also assigned with different set of <topic, partition>. To get unique transaction id,
     * one way is to append connector name, hostname and string of each <topic, partition> pair
     */
    protected String  sendBatchgetTransactionId(records, producer);) {
    	return getHostName() + "-" }+ catch getUniquePredictableStr(RebalanceException e) {;
    }

    @Override
	public void put(Collection<SinkRecord>  producer.close();records) {
        log.info("receive {} messages from producer = initProducer(isTransactional);  consumer", records.size());
        } catch (ResendRecordsException eif (records.size() == 0) {
        	return;
    abortTransaction(producer);
    }
        //TODO: add limited retrytry {
            sendBatch(e.getRemainingRecords()records, producer);
        } catch (ThrowableRebalanceException e) {
            log.error(getHostName() + " terminating on exception: {}", e producer.close();
            producer = return;initProducer(isTransactional);  
        }
 catch (ResendRecordsException e) }{

      private void sendBatch(Collection<SinkRecord> records, KafkaProducer<byte[], byte[]> abortTransaction(producer);
  {
        try {
 //TODO: add limited retry
        Map<TopicPartition, List<SinkRecord> remainingRecordsMap = new HashMap<>( sendBatch(e.getRemainingRecords(), producer);
        } catch (Throwable  offsetsMap.clear();e) {
            beginTransactionlog.error(getHostName(producer);
 + " terminating on exception:       SinkRecord record{}", e);
            for ((record = records.peek()) != null) {
return;
        }
    }

    private void sendBatch(Collection<SinkRecord> records, ProducerRecord<byteKafkaProducer<byte[], byte[]> producerRecord = convertToProducerRecord(record);producer) {
        try {
       offsetsMap.compute(new TopicPartition(record.topic(), record.kafkaPartition()),
   Map<TopicPartition, List<SinkRecord> remainingRecordsMap = new HashMap<>();
            offsetsMap.clear();
   (tp, curOffsetMetadata) ->
         beginTransaction(producer);
            SinkRecord record;
         (curOffsetMetadata == null ||for ((record = records.kafkaOffsetpeek()) > curOffsetMetadata.offset())!= null) {
                ProducerRecord<byte[], byte[]> producerRecord = convertToProducerRecord(record);
                offsetsMap.compute(new    ?TopicPartition(record.topic(), record.kafkaPartition()),
                       (tp, curOffsetMetadata) ->
              new OffsetAndMetadata(record.kafkaOffset())
                (curOffsetMetadata == null || record.kafkaOffset() > curOffsetMetadata.offset())
                 : curOffsetMetadata);
                Future<RecordMetadata> future = producer.send(producerRecord, (recordMetadata, e) -> {?
                    if (e != null) {
                       new logOffsetAndMetadata(record.error("{} failed to send record to {}: ", MirrorSinkTask.this, producerRecord.topic(), e);
kafkaOffset())
                                   log.debug("{} Failed record: {}", MirrorSinkTask.this, producerRecord : curOffsetMetadata);
                Future<RecordMetadata> future       throw new KafkaException(e);
= producer.send(producerRecord, (recordMetadata, e) -> {
                    if (e }!= elsenull) {
                        log.infoerror("{} Wrotefailed recordto successfully:send topicrecord {}to partition {}: offset {}", //log.trace", MirrorSinkTask.this, producerRecord.topic(), e);
                        log.debug("{} Failed record: {}", 		MirrorSinkTask.this, producerRecord);
                        throw new KafkaException(e);
      recordMetadata.topic(), recordMetadata.partition(),
             } else {
                 recordMetadata.offset());
       log.info("{} Wrote record successfully: topic {} partition {} offset      {}", //log.trace
   commitRecord(record, recordMetadata);
                    }		MirrorSinkTask.this,
                });
                futures.add(future);recordMetadata.topic(), recordMetadata.partition(),
                records.poll();
            }

        } catch (KafkaException e) {
recordMetadata.offset());
              // Any unsent messages are added to the remaining remainingRecordsMap for re-sendcommitRecord(record, recordMetadata);
            for (SinkRecord record = records.poll(); record != null; record}
 = records.poll()) {
            	addConsumerRecordToTopicPartitionRecordsMap(record, remainingRecordsMap});
          } finally {  //TODO: may futures.add(future);
 more exception handling case
            for (Future<RecordMetadata> future : futures) {
records.poll();
            }

      try {
 } catch (KafkaException e) {
            // Any  future.get();
        unsent messages are added to the remaining remainingRecordsMap for re-send
        } catch (Exception e) {
for (SinkRecord record = records.poll(); record != null; record = records.poll()) {
         SinkRecord record = futureMap.get(future	addConsumerRecordToTopicPartitionRecordsMap(record, remainingRecordsMap);
        } finally {  //TODO: may add more exception handling case
    // Any record failed to send, add to thefor remainingRecordsMap
(Future<RecordMetadata> future : futures) {
                addConsumerRecordToTopicPartitionRecordsMap(record, remainingRecordsMap);try {
                }
    future.get();
        }
         }

 catch  (Exception e) {
    if (isTransactional && remainingRecordsMap.size() == 0) {
          SinkRecord record = producerfutureMap.sendOffsetsToTransaction(offsetsMap, consumerGroupIdget(future);
             commitTransaction(producer);
       // Any record failed to }

send, add to the remainingRecordsMap
    if (remainingRecordsMap.size() > 0) {
            // For transaction case, all records should be put into remainingRecords, as the whole transaction should be redoneaddConsumerRecordToTopicPartitionRecordsMap(record, remainingRecordsMap);
                }
            Collection<SinkRecord> recordsToReSend;}
         }

   if (isTransactional) {
    if (isTransactional && remainingRecordsMap.size() == 0) {
      // transactional: retry all records, as the transaction will cancel all successful and failed records.
 producer.sendOffsetsToTransaction(offsetsMap, consumerGroupId);
             commitTransaction(producer);
       recordsToReSend = records;}

        if (remainingRecordsMap.size() > 0) } else {
            // For transaction  // non-transactional: only retry failed records, others were finished and sent.
case, all records should be put into remainingRecords, as the whole transaction should be redone
            Collection<SinkRecord> recordsToReSend;
       recordsToReSend = remainingRecordsMap;
   if (isTransactional) {
       }
         // transactional: retry throwall new ResendRecordsException(recordsToReSend);
        }records, as the transaction will cancel all successful and failed records.
    }

    @Override
    public Map<TopicPartition, OffsetAndMetadata> preCommit(Map<TopicPartition, OffsetAndMetadata>recordsToReSend offsets)= {records;
    	// if transactional, return empty Map intentionally to disable the offset commit by commitOffsets() in WorkerSinkTask.java
        } else {
                // non-transactional: soonly thatretry thefailed transactionalrecords, producerothers iswere ablefinished toand commitsent.
 the consumer offsets to target cluster in a transaction at its pace
    	if (isTransactional) recordsToReSend = remainingRecordsMap;
    		return new HashMap<TopicPartition, OffsetAndMetadata>();
        }
      	else  // otherwise, return offsetsMap tothrow letnew commitOffsetsResendRecordsException(recordsToReSend);
 in "WorkerSinkTask.java" to commit offsets
    		return offsetsMap;	}
    }

    // This commitRecord() follows the same logics as commitRecord() in MirrorSourceTask, to 
    public void commitRecord(SinkRecord record, RecordMetadata metadata) {
        try {
            if (stopping) {
                return;
            }
            if (!metadata.hasOffset()) {
                log.error("RecordMetadata has no offset -- can't sync offsets for {}.", record.topic());
                return;
            }
            TopicPartition topicPartition = new TopicPartition(record.topic(), record.kafkaPartition());
            long latency = System.currentTimeMillis() - record.timestamp();
            metrics.countRecord(topicPartition);
            metrics.replicationLatency(topicPartition, latency);
            TopicPartition sourceTopicPartition = MirrorUtils.unwrapPartition(record.sourcePartition());
            long upstreamOffset = MirrorUtils.unwrapOffset(record.sourceOffset());
            long downstreamOffset = metadata.offset();
            maybeSyncOffsets(sourceTopicPartition, upstreamOffset, downstreamOffset);
        } catch (Throwable e) {
            log.warn("Failure committing record.", e);
        }
    }

    private void beginTransaction(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional) {
            producer.beginTransaction();
            transactionStarttransactionInProgress = true;
        }
    }
	
    private void initTransactions(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional) {
            producer.initTransactions();
        }
    }
    
    private void commitTransaction(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional) {
            producer.commitTransaction();
            transactionStarttransactionInProgress = false;
        }
    }
    
    private void abortTransaction(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional && transactionStarttransactionInProgress) {
            producer.abortTransaction();
            transactionStarttransactionInProgress = false;
        }
    }

    public static class ResendRecordsException extends Exception {
        private Collection<SinkRecord> remainingRecords;

        public ResendRecordsException(Collection<SinkRecord> remainingRecords) {
            super(cause);
            this.remainingRecords = remainingRecords;
        }

        public Collection<SinkRecord> getRemainingRecords() {
            return remainingRecords;
        }
    }