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) {
			Map<TopicPartition, OffsetAndMetadata> initOffsetsOnTarget = listTargetConsumerGroupOffsets(connectConsumerGroup);
	
			// To rewind consumer to correct offset per <topic, partition> assigned to this task, there are more than one case:        loadContextOffsets();
        }
    }

    @Override
    public void open(Collection<TopicPartition>      // (1) offset exists on target, use target offset as the starting point of consumptionpartitions) {
		if (isTransactional) {
    	    loadContextOffsets();
		}
    }

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

 offset DOES NOT exist on target, presume a// newonly topic,keep sothe setoffsets offsetof tothe 0
partitions assigned to this task
        Map<TopicPartition, Long> taskOffsetscontextOffsets = taskTopicPartitionsassignments.stream().collect(Collectors.toMap(

        		        		x -> x, x -> initOffsetsOnTarget.containsKey(x) 
            		?
                    initOffsetsOnTarget.getfilter(x).offset() + 1
 -> currentOffsets.containsKey(x))
                       : 0L));
			context.offset(taskOffsets);
        }
    }

	protected KafkaProducer<byte[], byte[]> initProducer(boolean isTransactional) {
        Map<String, Object> producerConfig = config.targetProducerConfig();
        if (isTransactional) {.collect(Collectors.toMap(
        	String	  transactionId = getTransactionId();
        	log.info("use transactional producer with Id: {} ", transactionId);
            producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
            producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
 x -> x, x    	producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);-> currentOffsets.get(x)));

        	producerConfigcontext.put(ProducerConfig.CLIENT_ID_CONFIG, transactionIdoffset(contextOffsets);
        }

	protected KafkaProducer<byte[], byte[]> initProducer(boolean isTransactional) {
        Map<String, Object> producerConfig return= MirrorUtilsconfig.newProducertargetProducerConfig(producerConfig);
	}

    /**
    if * Per some articles, to avoid ProducerFencedException, transaction id is suggested to set application name + hostname
     * Each MirrorSinkTask is also assigned with different set of <topic, partition>. To get unique transaction id,(isTransactional) {
        	String transactionId = getTransactionId();
        	log.info("use transactional producer with Id: {} ", transactionId);
            producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
     * one way is to append connector name, hostname and string of each <topic, partition> pair
     */
 producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        	producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);
      protected String getTransactionId() {
	producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, transactionId);
        	return}
 connectorName + "-" + getHostName() + "-" +return getTopicPartitionStrMirrorUtils.newProducer(producerConfig);
	}

    }/**
    
  * Per protected String getTopicPartitionStr() {
    	StringBuffer buf = new StringBuffer();
    	for (TopicPartition tp :  taskTopicPartitions) {
    		buf.append(tp.topic() + "_" + tp.partition() + "-");
    	}
    	return buf.toString().substring(0, buf.length() - 1);
    }

    @Override
	public void put(Collection<SinkRecord> recordssome articles, to avoid ProducerFencedException, transaction id is suggested to set application name + hostname
     * Each MirrorSinkTask 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 getTransactionId() {
    	return getHostName() + "-" + getUniquePredictableStr();
    }

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

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

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

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

    // @Override
This commitRecord() follows the publicsame Map<TopicPartition,logics OffsetAndMetadata>as preCommit(Map<TopicPartitioncommitRecord() in MirrorSourceTask, OffsetAndMetadata>to offsets) {
    	// if transactional, return empty Map intentionally to disable the offset commit by commitOffsets() in WorkerSinkTask.java
public void commitRecord(SinkRecord record, RecordMetadata metadata) {
        try {
         // so that theif transactional producer is able to commit the consumer offsets to target cluster in a transaction at its pace(stopping) {
                return;
    	if (isTransactional) 
    		return new HashMap<TopicPartition, OffsetAndMetadata>(); }
    	else // otherwise, return offsetsMap to let commitOffsets() in "WorkerSinkTask.java" to commit offsets
if (!metadata.hasOffset()) {
     		return offsetsMap;	
    }

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

 sourceTopicPartition = MirrorUtils.unwrapPartition(record.sourcePartition());
       private void beginTransaction(KafkaProducer<byte[], byte[]> producer) {
      long upstreamOffset =if MirrorUtils.unwrapOffset(record.sourceOffset());(isTransactional) {
            long downstreamOffset = metadata.offsetproducer.beginTransaction();
            maybeSyncOffsets(sourceTopicPartition, upstreamOffset, downstreamOffset)transactionInProgress = true;
        }
 catch (Throwable e) {
   }
	
    private void initTransactions(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional) {
          log.warn("Failure committing record.", eproducer.initTransactions();
        }
    }
    
    private void beginTransactioncommitTransaction(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional) {
            producer.beginTransactioncommitTransaction();
            transactionInProgress = truefalse;
        }
    }
	    
    private void initTransactionsabortTransaction(KafkaProducer<byte[], byte[]> producer) {
        if (isTransactional && transactionInProgress) {
            producer.initTransactionsabortTransaction();
            transactionInProgress = false;
        }
    }

    public static class ResendRecordsException extends Exception {
    private  void commitTransaction(KafkaProducer<byte[], byte[]> producer) { private Collection<SinkRecord> remainingRecords;

        ifpublic ResendRecordsException(isTransactionalCollection<SinkRecord> remainingRecords) {
            producer.commitTransactionsuper(cause);
            transactionInProgressthis.remainingRecords = falseremainingRecords;
        }

    }
    
public Collection<SinkRecord>   private void abortTransaction(KafkaProducer<byte[], byte[]> producer) getRemainingRecords() {
        if (isTransactional && transactionInProgress) {
            producer.abortTransaction();
            transactionInProgress = falsereturn remainingRecords;
        }
    }

    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;
        }
    }