Configuring the Mirus quickstart to replicate from one system to another

[marcaguzman]

I'm new to Kafka and Mirus and have been experimenting with the quickstarts for both of them. It was pretty easy to get the default configuration going (hats off to the Mirus team) on a Centos VM, but now I'd to like to set up replication from one instance of Kafka to another that is on a separate VM. Changing destination.bootstrap.servers to the IP of second VM had no effect. Assuming this is possible with the Mirus quickstart can someone advise on my next steps? Thank you.

[dalassi1]

Hi @marcaguzman , can you post your configs related to this issue to better help you? (the ones that worked and the ones that didn't)

[marcaguzman]

Wow! Thanks for such a fast reply @dalassi1. I used the default config initially with a few benign modifications to ensure I knew which kafka instance Mirus was connecting to because at one point I had Mirus running on both VMs (both in loopback), but now I have only one instance running. Mostly default loopback quickstart worker properties file:
`bootstrap.servers=localhost:9092

group.id=mirus-quickstart

key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
header.converter=org.apache.kafka.connect.converters.ByteArrayConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

config.storage.topic=mg2-config
status.storage.topic=mg2-status
offset.storage.topic=mg2-offsets

config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1`

Loopback connection config is the default:
> curl localhost:8083/connectors/mirus-quickstart-source/config \ -X PUT \ -H 'Content-Type: application/json' \ -d '{ "name": "mirus-quickstart-source", "connector.class": "com.salesforce.mirus.MirusSourceConnector", "tasks.max": "5", "topics.whitelist": "test", "destination.topic.name.suffix": ".mirror", "destination.bootstrap.servers": "localhost:9092", "consumer.bootstrap.servers": "localhost:9092", "consumer.client.id": "mirus-quickstart", "consumer.key.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer", "consumer.value.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer" }'

I made no additional changes to the worker file above, but did set destination.bootstrap.servers and name fields in the connection setup in my attempt to replicate to a separate system:
curl -o - localhost:8083/connectors/MIRUS-quickstart-source/config\ -X PUT \ -H 'Content-Type: application/json' \ -d '{ "name": "MIRUS-quickstart-source", "connector.class": "com.salesforce.mirus.MirusSourceConnector", "tasks.max": "5", "topics.whitelist": "test", "destination.topic.name.suffix": ".mirror", "destination.bootstrap.servers": "172.29.82.125:9092", "consumer.bootstrap.servers": "localhost:9092", "consumer.client.id": "MIRUS-quickstart", "consumer.key.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer", "consumer.value.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer" }'

[dalassi1]

This config looks right to me. Anything interesting in the logs? Also, did you make sure you have these configs right in the worker config file https://github.com/salesforce/mirus#mirus-specific-configuration ? Specifically, I can think of producer.bootstrap.servers=172.29.82.125:9092

[dalassi1]

This is an example of what producer configs you might need to have in your worker config file

producer.bootstrap.servers=<server-list>:<port-num>
producer.compression.type=gzip

[marcaguzman]

The log dir (mirus-master/quickstart/logs/) is empty. I'll restart Mirus with the addition of the producer configuration and tee the output .

[marcaguzman]

Mirus, with the producer controls added to the worker config, appears to start cleanly, but replication is not occurring between the two systems (separate VMs with kafka running on both). No errors in the Mirus logs. It's a pull configuration with Mirus running on the sink system which is supposed to receive messages posted in the source system. I'll start from zero tomorrow and if it still does not work will post the config files, connector creation curl, and logs. Thanks.

[marcaguzman]

No luck. I deleted the kafka logs in both nodes (Hub, GP), added the provider config lines, and ran through the setup steps. Msgs are not replicating from the Hub test topic to the GP test.mirror topic. Worker config file:

##
# Mirus Quickstart worker properties file
#
# This file can be passed to `bin/mirus-worker-start.sh` to quickly get a 
# basic Mirus worker up and running.
#
# ---
# Kafka broker bootstrap server - Source cluster is on a remote machine
bootstrap.servers=172.29.82.125:9092

group.id=mirus-quickstart

key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
header.converter=org.apache.kafka.connect.converters.ByteArrayConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

config.storage.topic=gp-pull-config
status.storage.topic=gp-pull-status
offset.storage.topic=gp-pull-offsets

# In a production environment these values should be increased
config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1

# Recommended by Derar Alassi [mailto:[email protected]] 
# This Mirus instance is on the same VM as the instance of Kafka the will 
# receive the replicated messages.
producer.bootstrap.servers=localhost:9092
producer.client.id=replikator-sink
producer.compression.type=gzip
producer.security.protocol=SSL

Connector curl:

curl localhost:8083/connectors/mirus-quickstart-source/config \
      -X PUT \
      -H 'Content-Type: application/json' \
      -d '{
           "name": "mirus-quickstart-source",
           "connector.class": "com.salesforce.mirus.MirusSourceConnector",
           "tasks.max": "5",
           "topics.whitelist": "test",
           "destination.topic.name.suffix": ".mirror",
           "destination.bootstrap.servers": "localhost:9092",
           "consumer.bootstrap.servers": "172.29.82.125:9092",
           "consumer.client.id": "mirus-quickstart",
           "consumer.key.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer",
           "consumer.value.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer"
       }'	

Mirus log:
mirus.log

One odd thing - when I created the connector the three storage config topics (gp-pull-*) appeared in both kafka instances.