private boolean isTransactional = config.getTransactionalProducer();
    private boolean transactionStart = false;
	protected Map<TopicPartition, OffsetAndMetadata> offsetsMap = new HashMap<>();

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

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

    private@Override
    public void sendBatchopen(Collection<SinkRecord>Collection<TopicPartition> records, KafkaProducer<byte[], byte[]> producerpartitions) {
		if (isTransactional) {
    	    try {loadContextOffsets();
		}
    }

    private void   loadContextOffsets() {
		Map<TopicPartition, List<SinkRecord>OffsetAndMetadata> remainingRecordsMapinitOffsetsOnTarget = new HashMap<>listTargetConsumerGroupOffsets(connectConsumerGroup);
	
        Set<TopicPartition> assignments =  offsetsMapcontext.clearassignment();

        // only keep the offsets  beginTransaction(producer);
     of the partitions assigned to this task
       SinkRecord record;
Map<TopicPartition, Long> contextOffsets = assignments.stream()
        for		 ((record = records.peek()) != null) {
                ProducerRecord<byte[], byte[]> producerRecord = convertToProducerRecord(record);
                     offsetsMap.computefilter(new TopicPartition(record.topic(), record.kafkaPartition()),x -> currentOffsets.containsKey(x))
                       (tp, curOffsetMetadata) ->
                                   (curOffsetMetadata == null || record.kafkaOffset() > curOffsetMetadata.offset())collect(Collectors.toMap(
        		                               ?
               x -> x, x -> currentOffsets.get(x)));

                   new OffsetAndMetadata(record.kafkaOffset())context.offset(contextOffsets);
    }

	protected KafkaProducer<byte[], byte[]> initProducer(boolean isTransactional) {
        Map<String, Object> producerConfig = config.targetProducerConfig();
        if (isTransactional) {
        : curOffsetMetadata	String transactionId = getTransactionId();
        	log.info("use transactional producer with Id:    Future<RecordMetadata> future = producer.send(producerRecord, (recordMetadata, e) -> {
{} ", transactionId);
            producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
         if (e != null) {
 producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        	producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);
        	producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, transactionId);
        log.error("{}
 failed to send record to {}: ", MirrorSinkTask.this, producerRecord.topic(), e);
 return MirrorUtils.newProducer(producerConfig);
	}

    /**
     * Per some articles, to avoid ProducerFencedException, transaction id is suggested to set application  log.debug("{} Failed record: {}", MirrorSinkTask.this, producerRecord);
                name + hostname
     * Each MirrorSinkTask is also assigned with different set of <topic, partition>. To get unique transaction id,
     * one way throwis new KafkaException(e);
          to append connector name, hostname and string of each <topic, partition> pair
     */
    protected }String elsegetTransactionId() {
    	return getHostName() + "-" + getUniquePredictableStr();
    }

    @Override
	public void put(Collection<SinkRecord> records) {
        log.info("{} Wrote record successfully: topic receive {} partitionmessages {} offset {}from consumer", //log.tracerecords.size());
        if (records.size() == 0) {
            		MirrorSinkTask.this,	return;
        }
        try {
               recordMetadata.topic(), recordMetadata.partition(),
sendBatch(records, producer);
        } catch (RebalanceException e) {
            producer.close();
            producer = recordMetadata.offsetinitProducer(isTransactional));  
        } catch (ResendRecordsException e) {
            commitRecord(record, recordMetadataabortTransaction(producer);
            //TODO: add limited retry
     }
       sendBatch(e.getRemainingRecords(), producer);
        });
 catch (Throwable e) {
            futureslog.adderror(getHostName(future);
 + " terminating on exception:           records.poll({}", e);
            }
return;
        }
 catch (KafkaException e) {}

    private void sendBatch(Collection<SinkRecord> records, KafkaProducer<byte[], byte[]>   // Any unsent messages are added to the remaining remainingRecordsMap for re-sendproducer) {
        try {
            forMap<TopicPartition, (SinkRecordList<SinkRecord> recordremainingRecordsMap = new records.pollHashMap<>(); record != null; record = records.poll()) {
            	addConsumerRecordToTopicPartitionRecordsMap(record, remainingRecordsMapoffsetsMap.clear();
        } finally {  //TODO: may add more exception handling case  beginTransaction(producer);
            for (Future<RecordMetadata> future : futures) {
SinkRecord record;
            for ((record = records.peek())     try!= null) {
                    future.get(ProducerRecord<byte[], byte[]> producerRecord = convertToProducerRecord(record);
                } catch (Exception e) {offsetsMap.compute(new TopicPartition(record.topic(), record.kafkaPartition()),
                    SinkRecord record = futureMap.get(future);(tp, curOffsetMetadata) ->
                    // Any record failed to send, add to the remainingRecordsMap
   (curOffsetMetadata == null || record.kafkaOffset() > curOffsetMetadata.offset())
             addConsumerRecordToTopicPartitionRecordsMap(record, remainingRecordsMap);
                }
         ?
   }
         }

         if (isTransactional && remainingRecordsMap.size() == 0) {
            new producerOffsetAndMetadata(record.sendOffsetsToTransaction(offsetsMap, consumerGroupId);
kafkaOffset())
              commitTransaction(producer);
         }

        if (remainingRecordsMap.size() > 0) {
    : curOffsetMetadata);
       // For transaction case, all records should be put intoFuture<RecordMetadata> remainingRecords,future as the whole transaction should be redone= producer.send(producerRecord, (recordMetadata, e) -> {
            Collection<SinkRecord> recordsToReSend;
       if (e    if (isTransactional!= null) {
                // Transactional: retry all records, the transaction will have cancelled all of our outputs.
      log.error("{} failed to send record to {}: ", MirrorSinkTask.this, producerRecord.topic(), e);
           recordsToReSend = records;
            } else {
    log.debug("{} Failed record: {}", MirrorSinkTask.this, producerRecord);
            // Non-transactional: only retry failed records, others already were finished and sent.
 throw new KafkaException(e);
             recordsToReSend = remainingRecordsMap;
     } else {
     }
            throw new ResendRecordsException(recordsToReSend);
     log.info("{} Wrote record }
successfully: topic {} partition {}

 offset {}", //log.trace
    @Override
    public Map<TopicPartition, OffsetAndMetadata> preCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
    	// if transactional, return empty Map intentionally to disable the offset commit by commitOffsets() in "WorkerSinkTask.java"
 		MirrorSinkTask.this,
                 // so that the transactional producer is able to commit the  consumer offsets in a transaction recordMetadata.topic(), recordMetadata.partition(),
    	if (isTransactional) 
    		return new HashMap<TopicPartition, OffsetAndMetadata>();
    	else // otherwise, return offsetsMap to let commitOffsets() in "WorkerSinkTask.java" to commit offsets
    		return offsetsMap;	 recordMetadata.offset());
    }

    // This commitRecord() follows the same logics as commitRecord() in MirrorSourceTask
    public void commitRecord(SinkRecord record, RecordMetadata metadatarecordMetadata);
 {
        try {
          }
  if (stopping) {
            });
    return;
            }futures.add(future);
            if (!metadata.hasOffset()) {    records.poll();
            }

    log.error("RecordMetadata has no offset -- can't sync offsets for {}.", record.topic());} catch (KafkaException e) {
            // Any unsent messages return;
are added to the remaining remainingRecordsMap for re-send
     }
       for (SinkRecord record   TopicPartition topicPartition = new TopicPartition(record.topic(), record.kafkaPartition());records.poll(); record != null; record = records.poll()) {
            long latency = System.currentTimeMillis() - record.timestamp(	addConsumerRecordToTopicPartitionRecordsMap(record, remainingRecordsMap);
        } finally {   metrics.countRecord(topicPartition);
     //TODO: may add more exception handling case
       metrics.replicationLatency(topicPartition, latency);
     for (Future<RecordMetadata> future : futures) {
    TopicPartition sourceTopicPartition = MirrorUtils.unwrapPartition(record.sourcePartition());
         try {
  long upstreamOffset = MirrorUtils.unwrapOffset(record.sourceOffset());
            long downstreamOffset = metadatafuture.offsetget();
            maybeSyncOffsets(sourceTopicPartition, upstreamOffset, downstreamOffset);
  } catch (Exception e) {
  }  catch (Throwable e) {
            log.warn("Failure committing SinkRecord record.", e = futureMap.get(future);
            }
       }

 // Any record privatefailed void beginTransaction(KafkaProducer<byte[], byte[]> producer) {to send, add to the remainingRecordsMap
        if (isTransactional) {
            producer.beginTransaction(addConsumerRecordToTopicPartitionRecordsMap(record, remainingRecordsMap);
              transactionStart = true;}
        }
    }
	
    private void initTransactions(KafkaProducer<byte[], byte[]> producer) {}

         if (isTransactional && remainingRecordsMap.size() == 0) {
             producer.initTransactionssendOffsetsToTransaction(offsetsMap, consumerGroupId);
        }
    }
     commitTransaction(producer);
    private void commitTransaction(KafkaProducer<byte[], byte[]> producer) {}

        if (isTransactionalremainingRecordsMap.size() > 0) {
            producer.commitTransaction();
            transactionStart = false;
   // For transaction case, all records should be put into remainingRecords, as the whole transaction should be redone
     }
    }
   Collection<SinkRecord> recordsToReSend;
    private      void abortTransaction(KafkaProducer<byte[], byte[]> producer if (isTransactional) {
              if (isTransactional && transactionStart) {
            producer.abortTransaction();
// transactional: retry all records, as the transaction will cancel all successful and failed records.
                transactionStartrecordsToReSend = falserecords;
        }
    }

    public static class ResendRecordsException extends Exception {
 else {
                // non-transactional: only retry failed records, others were finished and sent.
         private Collection<SinkRecord> remainingRecords;

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

        public Collection<SinkRecord> getRemainingRecords() {
            return remainingRecords;
        }
    }throw new ResendRecordsException(recordsToReSend);
        }
    }

    // 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();
            transactionInProgress = true;
        }
    }
	
    private void initTransactions(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional) {
            producer.initTransactions();
        }
    }
    
    private void commitTransaction(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional) {
            producer.commitTransaction();
            transactionInProgress = false;
        }
    }
    
    private void abortTransaction(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional && transactionInProgress) {
            producer.abortTransaction();
            transactionInProgress = 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;
        }
    }