Copy Topics Between ServiceNow Hermes and Confluent Cloud with Confluent Replicator

Adam Celli · 58m ago

This guide shows you, step by step, how to copy messages between ServiceNow Hermes and Confluent Cloud with Confluent Replicator, in both directions:

Part 1 (Steps 1.x): Hermes into Confluent Cloud. Replicator runs inside Confluent Cloud as a custom connector, so there is nothing to host.
Part 2 (Steps 2.x): Confluent Cloud into Hermes. Replicator runs in one small container that you host.

Both parts need the same Hermes certificate files, which you create in Step 1.1. If you only need Part 2, do Step 1.1 first, then skip to Step 2.1.

Along the way it lists the places where the setup differs from Confluent's own documentation, so you can skip the trial and error.

I tested every command in bash (Linux, macOS or WSL on Windows), using the Confluent CLI, Docker and curl. Where a step mentions the Confluent Cloud console, the console path follows Confluent's documentation.

A few terms

Hermes is ServiceNow's Kafka-based messaging service. Your instance publishes messages to it.
A topic is a named stream of messages, like a mailbox that producers write to and consumers read from.
Confluent Replicator is a Confluent component that copies messages from one Kafka cluster to another.
A custom connector is a connector you upload to Confluent Cloud yourself, so Confluent runs it for you.
A consumer group is the name a reader uses so Kafka can remember how far it has read.

Placeholders used throughout:

<env-id> and <lkc-id> are your Confluent Cloud environment and cluster IDs (confluent environment list, confluent kafka cluster list).
<instance> is your ServiceNow instance name, and <namespace> and <topic> are the Hermes namespace and topic.
<hermes-host> is your Hermes broker host name.
<api-key> and <api-secret> are your Confluent Cloud Kafka API key and secret.
<cc-bootstrap> and <jaas-config> are your Confluent Cloud bootstrap address and login string. Step 1.4 shows how to build them.
<source-topic> is the topic you copy from, <partitions> is its partition count, and <connector-id> is the ID that confluent connect cluster create returns.

Part 1: Copy Hermes into Confluent Cloud

Steps in this part are numbered 1.1 to 1.7. Step 1.1 (getting the PEM files) is also the starting point for Part 2.

What is different from the Confluent documentation

Confluent's Replicator docs assume you run Replicator on servers you manage. This guide runs it as a Confluent Cloud custom connector instead, so a few settings change:

Upload Replicator as a plugin (Step 1.2). Confluent Cloud runs it for you, so there is no server to set up.
Paste certificates in as text (Steps 1.1 and 1.4). A custom connector has no files to point at. Use Kafka's standard PEM properties (ssl.keystore.type=PEM and friends), and write each line break in a certificate as a literal backslash followed by n.
Use an unencrypted private key (Step 1.1). The connector settings have no place for a key password.
Add the connector's Confluent Cloud credentials (Step 1.4). Set kafka.auth.mode, kafka.api.key and kafka.api.secret.
**Leave out confluent.topic.replication.factor (Step 1.4).** Confluent Cloud sets it for you.
Name the Confluent Cloud destination explicitly (Step 1.4). Set the dest.kafka.* properties yourself instead of expecting Confluent Cloud to fill them in.
**Set offset.timestamps.commit to false (Step 1.4).** This stops Replicator from trying to create an extra bookkeeping topic on Hermes.
**Start the consumer group ID with snc.<instance>. (Step 1.3).** Hermes only lets a reader use group names that start with your instance prefix.
Create the destination topic yourself first (Step 1.3). This avoids a mismatch between Hermes topic settings and Confluent Cloud limits.
**List the Hermes address in confluent.custom.connection.endpoints (Step 1.5).** That is enough to let the connector reach Hermes. Confluent does not support fixed egress IP addresses for custom connectors, so you cannot allow-list Confluent by IP on the Hermes side.
Expect about 8 minutes before the connector is ready (Step 1.6). That was true even on a Basic cluster.

Prerequisites

A Confluent Cloud environment and cluster. A Basic cluster worked.
A Kafka API key and secret for that cluster
The Confluent CLI, logged in (confluent login)
The Replicator package from Confluent, zipped with its files at the top level of the zip. It is a licensed component. Per Confluent's docs it stops after a 30-day trial without a license.
Your Hermes client certificate files: a PKCS12 keystore and truststore, or PEM files
Your Hermes instance name and a Hermes namespace (see Managing namespaces in Hermes)

Know the Hermes limits before you start: 2 MB maximum message size, 36 hours of retention, up to 32 partitions per topic, 960 partitions in total across all topics, and GZIP, LZ4 or no compression for produced messages. See Exploring Hermes Messaging Service.

Step 1.1: Get PEM files

Hermes checks a certificate from your side on every connection (this is called mTLS), so you need three PEM files: ca.pem, client-cert.pem and client-key.pem. If you only have a PKCS12 keystore and truststore, convert them with openssl. I ran these commands against a Hermes keystore and truststore and compared the results with another extraction tool: the same three CA certificates, the same two-certificate client chain and the same private key came out:

openssl pkcs12 -in truststore -nokeys -out ca.pem
openssl pkcs12 -in keystore -nokeys -out client-cert.pem
openssl pkcs12 -in keystore -nocerts -nodes -out client-key.pem

I used OpenSSL 3.5 and did not need the -legacy flag. Older keystores on OpenSSL 3 may.
openssl writes "Bag Attributes" lines above each block. Delete them.
The -nodes flag matters: it writes the key unencrypted (BEGIN PRIVATE KEY). Make sure yours does not say BEGIN ENCRYPTED PRIVATE KEY.

Check that the first line of client-key.pem is -----BEGIN PRIVATE KEY-----, not -----BEGIN ENCRYPTED PRIVATE KEY----- or a "Bag Attributes" line. An encrypted key will not work, because the connector settings have no place for a password. Treat the file as a secret, because it ends up in the connector settings.

Step 1.2: Upload the plugin

CLI (see the custom-plugin create command reference). Set --cloud to the cloud your cluster runs on; I tested aws:

confluent connect custom-plugin create replicator-plugin \
  --plugin-file ./confluent-replicator.zip \
  --connector-type source \
  --connector-class io.confluent.connect.replicator.ReplicatorSourceConnector \
  --sensitive-properties "src.kafka.sasl.jaas.config,dest.kafka.sasl.jaas.config,src.kafka.ssl.keystore.key" \
  --cloud aws

Console: open Connectors, add a custom connector plugin, upload the zip, set the class and type to the values above, and mark the same properties as sensitive.

Note the returned plugin ID (ccp-xxxxxx). The sensitive list only controls which values Confluent hides, so list the properties that really hold secrets. Plugins are not tied to a cluster. Confluent also documents a newer confluent ccpm plugin create flow in API and CLI for custom connectors. The command above uses the older custom-plugin path, which is what I tested.

Step 1.3: Choose topics, group ID and the destination topic

Use topic.whitelist for a comma-separated list of the Hermes topics to copy. Hermes topic names look like snc.<instance>.<namespace>.<topic>, and topic.rename.format sets the name of the copy on Confluent Cloud, for example replicator_test_b_${topic}. For the Hermes naming format, see Managing topics in Hermes.

Consumer group ID. Replicator reads from Hermes as a consumer group, for example snc.<instance>.replicator-<name>-4100. Use one group per Hermes cluster, and don't change it after the first run, because the group remembers how far Replicator has read.

This is the key point: the consumer group ID must start with snc.<instance>., or Hermes will not let Replicator read.

Create the destination topic first. Replicator would otherwise create it with settings copied from the Hermes topic, and Confluent Cloud rejects one of them (segment.bytes is above its limit). The name must be exactly what topic.rename.format produces. With the Step 1.4 config that is replicator_test_b_snc.<instance>.<namespace>.<topic>. This command creates it:

confluent kafka topic create replicator_test_b_snc.<instance>.<namespace>.<topic> \
  --partitions <partitions> --environment <env-id> --cluster <lkc-id>

Set <partitions> to the source topic's partition count.

Step 1.4: Build the connector config

This is the full connector.json I ran. It reads from the first Hermes cluster (ports 4100 to 4103). The PEM placeholders must be each file's contents, written on one line with \n where the line breaks were:

{
  "name": "hermes-to-cc",
  "config": {
    "name": "hermes-to-cc",
    "connector.class": "io.confluent.connect.replicator.ReplicatorSourceConnector",
    "confluent.custom.plugin.id": "ccp-xxxxxx",
    "confluent.connector.type": "CUSTOM",
    "kafka.auth.mode": "KAFKA_API_KEY",
    "kafka.api.key": "<api-key>",
    "kafka.api.secret": "<api-secret>",
    "tasks.max": "1",
    "offset.timestamps.commit": "false",
    "topic.config.sync": "false",
    "topic.whitelist": "snc.<instance>.<namespace>.<topic>",
    "topic.rename.format": "replicator_test_b_${topic}",
    "key.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
    "value.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
    "header.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
    "src.key.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
    "src.value.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
    "src.header.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
    "src.kafka.bootstrap.servers": "<hermes-host>:4100,<hermes-host>:4101,<hermes-host>:4102,<hermes-host>:4103",
    "src.kafka.security.protocol": "SSL",
    "src.kafka.ssl.truststore.type": "PEM",
    "src.kafka.ssl.truststore.certificates": "<ca.pem contents>",
    "src.kafka.ssl.keystore.type": "PEM",
    "src.kafka.ssl.keystore.certificate.chain": "<client-cert.pem contents>",
    "src.kafka.ssl.keystore.key": "<client-key.pem contents>",
    "src.consumer.group.id": "snc.<instance>.replicator-<name>-4100",
    "dest.kafka.bootstrap.servers": "<cc-bootstrap>",
    "dest.kafka.security.protocol": "SASL_SSL",
    "dest.kafka.sasl.mechanism": "PLAIN",
    "dest.kafka.sasl.jaas.config": "<jaas-config>",
    "confluent.custom.connection.endpoints": "<hermes-host>:4100,4101,4102,4103"
  }
}

Fill in the placeholders:

<hermes-host>: your Hermes broker host (in my test it had the form <instance>.service-now.com). The read ports are published in the Hermes docs. 4100 to 4103 is cluster 1 and 4200 to 4203 is cluster 2.
<cc-bootstrap>: run confluent kafka cluster describe <lkc-id> --environment <env-id> --output json and strip SASL_SSL:// from the endpoint.
<jaas-config>: org.apache.kafka.common.security.plain.PlainLoginModule required username="<api-key>" password="<api-secret>";. Inside JSON, escape each inner quote as \". In a .properties file, use it exactly as shown here.
<ca.pem contents> and the other two PEM placeholders: print each file as one line with \n in place of the line breaks, then paste the output:
```
python3 -c "import sys; print(open(sys.argv[1]).read().strip().replace(chr(10), chr(92) + 'n'))" ca.pem
```
<name>: any short label for this connector, such as prod.
The src.kafka.* block carries the mTLS settings for Hermes. Leave out confluent.topic.replication.factor, and don't add producer.override.bootstrap.servers or the producer.override.ssl.* keys. Confluent Cloud does not allow them on custom connectors.

Hermes is an active-active pair of clusters. For the second cluster, repeat the config with ports 4200 to 4203, a new connector name and group ID, and the matching endpoint. I tested cluster 1 only.

Step 1.5: Let the connector reach Hermes

confluent.custom.connection.endpoints in the config above lists the Hermes host and ports, and in my tests that was enough. If you still see connection errors, add the same endpoints in the connector's Networking settings in the console.

Confluent's endpoint requirements are:

Use a fully-qualified host name.
Separate multiple ports with commas and multiple endpoints with semicolons.
Do not include http:// or https://.

Step 1.6: Create the connector

CLI:

confluent connect cluster create --config-file connector.json \
  --environment <env-id> --cluster <lkc-id>

Console: Connectors, add the custom connector, pick your plugin, and paste the config.

Creation takes a while. Check the status until it says RUNNING, which took roughly 7 to 8 minutes for me:

confluent connect cluster describe <connector-id> --environment <env-id> --cluster <lkc-id>

If the status says FAILED, look in the connector's log topic, <connector-id>-app-logs, on the same cluster. Consume it and look for ERROR lines.

Step 1.7: Verify

Produce a test message to the Hermes source topic.
Consume from the destination topic on Confluent Cloud, for example with confluent kafka topic consume.
Confirm the message arrives. In my test, every message produced to Hermes arrived in order on the destination topic, including ones produced while the connector was already running.

The connector can say RUNNING even when no data is moving, so the consume check is the real test. Test topics you create on Hermes remain on both Hermes clusters. Deleting a topic only removes it from the cluster you deleted it on, and Hermes keeps protected copies that it cleans up on its own.

Part 2: Copy the other way, Confluent Cloud into Hermes

Steps in this part are numbered 2.1 to 2.6. They reuse the PEM files from Step 1.1.

This direction runs Replicator yourself in one small container, because Confluent Cloud allows only tuning overrides on custom connectors, so a custom connector's writes cannot be pointed at Hermes. On a Connect worker you run yourself, connector.client.config.override.policy=All allows it. The container's Connect worker keeps all of its bookkeeping on Confluent Cloud, and only the copied messages go to Hermes.

Confluent's Replicator docs usually place the Connect worker next to the destination cluster. Here it sits next to the source (Confluent Cloud) on purpose, because Hermes does not allow the worker's bookkeeping topics.

You also need Docker, and a host that can reach Confluent Cloud and the Hermes ports 4000 to 4003 (writes) and 4100 (verification reads).

How the pieces fit

The Connect worker connects to Confluent Cloud. Its config, offset and status topics are created there, and so are Replicator's _confluent-command license topic and __consumer_timestamps topic. Hermes refuses topics like these, so they must stay on Confluent Cloud.
Replicator reads the source topic from Confluent Cloud (src.kafka.*).
Replicator writes to Hermes through producer.override.*, and its admin client checks the Hermes topic through dest.kafka.*. Both use mTLS on the Hermes write ports (4000 to 4003 in my test).

I tested this with the apache/kafka:4.2.0 image (no Confluent Platform needed), Replicator 8.2.1 and the same PEM files from Step 1.1, against Confluent Cloud Basic and one Hermes cluster.

Step 2.1: Prepare a folder

In your working folder, create a folder named certs and put these files in it. The container mounts it as /certs:

ca.pem: from Step 1.1.
keystore.pem: client-key.pem and client-cert.pem joined into one file. The key must come first and must be unencrypted:
```
cat client-key.pem client-cert.pem > keystore.pem
```

hermes-client.properties: for the command-line tools.

security.protocol=SSL
ssl.keystore.type=PEM
ssl.keystore.location=/certs/keystore.pem
ssl.truststore.type=PEM
ssl.truststore.location=/certs/ca.pem

cc-client.properties: for the command-line tools.

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=<jaas-config>

worker.properties: shown in Step 2.2.

Unzip the Replicator package into a folder named confluent-replicator next to certs. The container mounts it as /plugins/replicator. Unlike Step 1.2, there is nothing to upload.

These files hold your private key and API secret. Keep them out of source control.

Step 2.2: Write the worker config

worker.properties points the worker at Confluent Cloud and turns on client overrides. <jaas-config> is the JAAS line from Step 1.4 in its plain form, without the JSON escaping. Step 2.5 is JSON, so there it needs the escaped form.

bootstrap.servers=<cc-bootstrap>
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=<jaas-config>
producer.security.protocol=SASL_SSL
producer.sasl.mechanism=PLAIN
producer.sasl.jaas.config=<jaas-config>
consumer.security.protocol=SASL_SSL
consumer.sasl.mechanism=PLAIN
consumer.sasl.jaas.config=<jaas-config>
admin.security.protocol=SASL_SSL
admin.sasl.mechanism=PLAIN
admin.sasl.jaas.config=<jaas-config>
group.id=replicator-cc-to-hermes
config.storage.topic=replicator-cc-to-hermes-configs
offset.storage.topic=replicator-cc-to-hermes-offsets
status.storage.topic=replicator-cc-to-hermes-status
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3
key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
listeners=http://0.0.0.0:8083
plugin.path=/plugins
connector.client.config.override.policy=All

The last line is what allows producer.override.*.

Step 2.3: Start the container

Run this from your working folder. It starts one Connect worker, mounts the certs and confluent-replicator folders read-only, and publishes the worker's REST port 8083 on local port 8085:

docker run -d --name replicator-connect --restart unless-stopped \
  -p 8085:8083 \
  -v "$PWD/certs":/certs:ro \
  -v "$PWD/certs/worker.properties":/config/worker.properties:ro \
  -v "$PWD/confluent-replicator":/plugins/replicator:ro \
  apache/kafka:4.2.0 \
  /opt/kafka/bin/connect-distributed.sh /config/worker.properties

After about 30 seconds, check that the worker loaded Replicator:

curl -s localhost:8085/connector-plugins | grep Replicator

You should see io.confluent.connect.replicator.ReplicatorSourceConnector.

Step 2.4: Create the topics

Create the Hermes destination topic yourself, with no topic configs, because Hermes rejects topics created with copied settings. Use the Hermes naming format snc.<instance>.<namespace>.<topic>:

docker exec replicator-connect /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server <hermes-host>:4000 \
  --command-config /certs/hermes-client.properties \
  --create --topic snc.<instance>.<namespace>.<topic> --partitions 1

If you don't already have a source topic on Confluent Cloud, create one (confluent kafka topic create <source-topic> --environment <env-id> --cluster <lkc-id>).

Step 2.5: Create the connector

Save this as cc-to-hermes.json. The ssl.* paths are inside the container, so unlike Step 1.4 you don't need to paste the PEM contents in.

{
  "connector.class": "io.confluent.connect.replicator.ReplicatorSourceConnector",
  "tasks.max": "1",
  "topic.whitelist": "<source-topic>",
  "topic.rename.format": "snc.<instance>.<namespace>.<topic>",
  "topic.auto.create": "false",
  "topic.config.sync": "false",
  "topic.preserve.partitions": "false",
  "offset.topic.commit": "false",
  "offset.translator.tasks.max": "0",
  "key.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
  "value.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
  "header.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
  "src.key.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
  "src.value.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",
  "src.header.converter": "io.confluent.connect.replicator.util.ByteArrayConverter",

  "src.kafka.bootstrap.servers": "<cc-bootstrap>",
  "src.kafka.security.protocol": "SASL_SSL",
  "src.kafka.sasl.mechanism": "PLAIN",
  "src.kafka.sasl.jaas.config": "<jaas-config>",
  "src.kafka.ssl.truststore.type": "JKS",
  "src.kafka.ssl.truststore.location": "/opt/java/openjdk/lib/security/cacerts",
  "src.consumer.group.id": "replicator-cc-to-hermes",

  "confluent.topic.bootstrap.servers": "<cc-bootstrap>",
  "confluent.topic.security.protocol": "SASL_SSL",
  "confluent.topic.sasl.mechanism": "PLAIN",
  "confluent.topic.sasl.jaas.config": "<jaas-config>",
  "confluent.topic.ssl.truststore.type": "JKS",
  "confluent.topic.ssl.truststore.location": "/opt/java/openjdk/lib/security/cacerts",

  "dest.kafka.bootstrap.servers": "<hermes-host>:4000,<hermes-host>:4001,<hermes-host>:4002,<hermes-host>:4003",
  "dest.kafka.security.protocol": "SSL",
  "dest.kafka.ssl.keystore.type": "PEM",
  "dest.kafka.ssl.keystore.location": "/certs/keystore.pem",
  "dest.kafka.ssl.truststore.type": "PEM",
  "dest.kafka.ssl.truststore.location": "/certs/ca.pem",

  "producer.override.bootstrap.servers": "<hermes-host>:4000,<hermes-host>:4001,<hermes-host>:4002,<hermes-host>:4003",
  "producer.override.security.protocol": "SSL",
  "producer.override.ssl.keystore.type": "PEM",
  "producer.override.ssl.keystore.location": "/certs/keystore.pem",
  "producer.override.ssl.truststore.type": "PEM",
  "producer.override.ssl.truststore.location": "/certs/ca.pem"
}

Then create it and check its status:

curl -s -X PUT -H "Content-Type: application/json" \
  --data @cc-to-hermes.json localhost:8085/connectors/cc-to-hermes/config
curl -s localhost:8085/connectors/cc-to-hermes/status

Unlike a custom connector, this one starts in seconds.

Why these settings matter:

**producer.override.*** is what sends the messages to Hermes. Without it, they go to Confluent Cloud.
**confluent.topic.*** keeps the license topic on Confluent Cloud.
**The two ssl.truststore settings that point at Java's cacerts** matter. Without them, Replicator's Confluent Cloud admin client picked up the Hermes truststore and failed with PKIX path building failed. Setting them makes the Confluent Cloud clients trust public certificates again.
**offset.topic.commit=false** (not the same property as offset.timestamps.commit in Step 1.4) is required when topic.preserve.partitions=false. Otherwise the connector fails with offset.topic.commit cannot be true if topic.preserve.partitions is false. It also stops Replicator from committing consumer offsets on Hermes, which it would not be allowed to do.
**offset.translator.tasks.max=0** turns off consumer offset translation, which would also write to Hermes.
**topic.auto.create=false and topic.config.sync=false** stop Replicator from creating or changing the Hermes topic.
**topic.rename.format** is set to the full Hermes topic name. With one source topic per connector, the name can be a fixed string.
**src.consumer.group.id** belongs to Confluent Cloud in this direction, so it doesn't need the snc. prefix.

Step 2.6: Verify

Produce to the Confluent Cloud source topic:

docker exec -i replicator-connect /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server <cc-bootstrap> --command-config /certs/cc-client.properties \
  --topic <source-topic>

Type a few test messages, one per line, then press Ctrl+C.

Consume from Hermes on a read port. The group name needs the snc.<instance>. prefix:

docker exec replicator-connect /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server <hermes-host>:4100 --command-config /certs/hermes-client.properties \
  --group snc.<instance>.replicator-verify \
  --topic snc.<instance>.<namespace>.<topic> --from-beginning --timeout-ms 20000

In my test, two messages produced to Confluent Cloud arrived on Hermes at offsets 0 and 1. No Hermes-named topic appeared on Confluent Cloud.

Also check that nothing went to the wrong place. Listing the Confluent Cloud topics (confluent kafka topic list --environment <env-id> --cluster <lkc-id>) should not show your Hermes topic name. You should see the worker's three replicator-cc-to-hermes-* topics, _confluent-command and __consumer_timestamps there. That is expected.

To check the connector itself, run docker logs replicator-connect. You can also remove the connector with curl -X DELETE localhost:8085/connectors/cc-to-hermes.

Related options

Cluster Linking is another way to bring Hermes topics into Confluent Cloud without a connector. It needs a cluster type that supports it (Basic clusters do not), and I will cover it in a separate article.