[KAFKA-14980] MirrorMaker consumers don't get configs prefixed with source.cluster

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 3.5.0
• Fix Version/s: 3.5.0
• Component/s: mirrormaker
• Assignee: Chris Egerton
• Reporter: Mickael Maison
• Created: 09/May/23 16:34
• Updated: 22/May/23 14:04
• Resolved: 19/May/23 14:04

Description

As part of KAFKA-14021, we made a change to MirrorConnectorConfig.sourceConsumerConfig() to grab all configs that start with "source.". Previously it was grabbing configs prefixed with "source.cluster.".

This means existing connector configuration stop working, as configurations such as bootstrap.servers are not passed to source consumers.

For example, the following connector configuration was valid in 3.4 and now makes the connector tasks fail:

{
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "name": "source",
    "topics": "test",
    "tasks.max": "30",
    "source.cluster.alias": "one",
    "target.cluster.alias": "two",
    "source.cluster.bootstrap.servers": "localhost:9092",
   "target.cluster.bootstrap.servers": "localhost:29092"
}

The connector attempts to start source consumers with bootstrap.servers = [] and the task crash with

org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:837)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:671)
	at org.apache.kafka.connect.mirror.MirrorUtils.newConsumer(MirrorUtils.java:59)
	at org.apache.kafka.connect.mirror.MirrorSourceTask.start(MirrorSourceTask.java:103)
	at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.initializeAndStart(AbstractWorkerSourceTask.java:274)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259)
	at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:75)
	at 
org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers

Issue Links

• links to: GitHub Pull Request #13723