Messages duplicated after container restart

Nobody has claimed this yet.

### Subject of the issue
I deploy Brooklin with Kubernetes to migrate data from one Kafka cluster to another.
For some reason the container dies and restart during the migration. At the end of the migration I noticed that there are more messages in the target cluster than there are in the source cluster. Most of the messages are duplicated.

I’d like to know how can I ensure that no message will be duplicated in the same data stream, even when the pod restarts after failure.

The datastream :

`
${BROOKLIN_HOME}/bin/brooklin-rest-client.sh -o CREATE -u http://localhost:32311/ -n test-mirroring-stream -s "kafkassl://kafka.kafka-non-prod:9092/.*” -c kafkaMirroringC -t kafkaTP -m '{"owner":"test-user","system.reuseExistingDestination":"false"}'
`

Config :
```
############################# Server Basics #############################
brooklin.server.coordinator.cluster=brooklin-cluster
brooklin.server.coordinator.zkAddress=zookeeper:2181/brooklin
brooklin.server.httpPort=${HTTP_PORT}
brooklin.server.connectorNames=kafkaC,kafkaMirroringC
brooklin.server.transportProviderNames=kafkaTP
brooklin.server.csvMetricsDir=/tmp/brooklin-example/

########################### Kafka connector Configs ######################
brooklin.server.connector.kafkaC.factoryClassName=com.linkedin.datastream.connectors.kafka.KafkaConnectorFactory
brooklin.server.connector.kafkaC.assignmentStrategyFactory=com.linkedin.datastream.server.assignment.BroadcastStrategyFactory
brooklin.server.connector.kafkaC.assignmentStrategyFactory=com.linkedin.datastream.server.assignment.BroadcastStrategyFactory
brooklin.server.connector.kafkaC.consummer.enable.auto.commit=false
brooklin.server.connector.kafkaC.consummer.auto.offset.reset=latest
brooklin.server.connector.kafkaC.retryCount=2147483647
brooklin.server.connector.kafkaC.retrySleepDurationMs=2500

########################### Kafka Mirroring connector Configs ###################### (Source properties)
brooklin.server.connector.kafkaMirroringC.factoryClassName=com.linkedin.datastream.connectors.kafka.mirrormaker.KafkaMirrorMakerConnectorFactory
brooklin.server.connector.kafkaMirroringC.assignmentStrategyFactory=com.linkedin.datastream.server.assignment.BroadcastStrategyFactory
brooklin.server.connector.kafkaMirroringC.isFlushlessModeEnabled=true
brooklin.server.connector.kafkaMirroringC.flowControlEnabled=true
brooklin.server.connector.kafkaMirroringC.maxInFlightMessagesThreshold=1
brooklin.server.connector.kafkaMirroringC.minInFlightMessagesThreshold=1

########################### Kafka transport provider configs ###################### (Target properties)
brooklin.server.transportProvider.kafkaTP.factoryClassName=com.linkedin.datastream.kafka.KafkaTransportProviderAdminFactory
brooklin.server.transportProvider.kafkaTP.bootstrap.servers=kafka-headless.da-saas-rss-env-region.svc.cluster.local:9092
brooklin.server.transportProvider.kafkaTP.zookeeper.connect=zookeeper:2181/kafka
brooklin.server.transportProvider.kafkaTP.client.id=datastream-producer
brooklin.server.transportProvider.kafkaTP.retries=2147483647
brooklin.server.transportProvider.kafkaTP.acks=all
...
```

### Your environment
* Operating System: Alpine
* Brooklin version: 1.0.2
* Java version: openjdk:8u212
* Kafka version: 2.5.0
* ZooKeeper version: 3.6.1
*
### Steps to reproduce
1. Create a datastream in a pod.
2. Restart the container in the pod.

### Expected behaviour
The datastream should resume without duplicating any message that has already been copied.

### Actual behaviour
The datastream doesn’t resume, most of the messages are duplicated.

Start with the Kafka mirroring connector configured through KafkaMirrorMakerConnectorFactory and reproduce the container restart described in the issue using the supplied datastream and Kafka settings. Trace how the stream resumes after failure and verify the result against the source and target message counts; done means the migration resumes without duplicating already-copied messages.

Written by the indexing model from the issue text.