THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
forwarding.admin.classdefault value will be set toForwardingAdmin with KafkaAdminClient as delegate.- or can be configured based on cluster aliases using
<cluster_alias>.cluster.forwarding.admin.class
The configuration for custom resource management client and/or KafkaAdminClient can be passed using the following prefix
admin.<any_config>*<cluster_alias>.cluster.admin.<any_config>*
Proposed Changes
ForwardingAdmin class
...
All Connectors and Tasks will replace AdminClient.create byMirrorConnectorConfig.forwardingAdminMirrorCheckpointTaskMirrorCheckpointConnectorMirrorSourceConnector
Use ForwardingAdmin in MirrorUtils instead of TopicAdmin to create internal compacted topics
Connector Configuration Properties
Properties common to the SourceConnector(s) and SinkConnector:
| property | default value | description | |
|---|---|---|---|
forwarding.admin.class | org.apache.kafka.clients.admin.ForwardingAdminClient | The fully qualified name of class that extend ForwardingAdminClient. The class must have a contractor that accept configuration (Map<String, Object> config) to configure needed clients. | |
target.cluster.forwarding.admin.class | org.apache.kafka.clients.admin.ForwardingAdminClient | Override | |
source.cluster.forwarding.admin.class | org.apache.kafka.clients.admin.ForwardingAdminClient | Override |
In addition, forwarding admin class will be re-using the following existing configs:
property | description |
| source.cluster.admin.* | overrides for the source-cluster forwarding admin |
| target.cluster.admin.* | overrides for the target-cluster forwarding admin |
Example Configuration
A sample configuration file ./config/connect-mirror-source.properties is provided for use case where source cluster use default org.apache.kafka.clients.admin.ForwardingAdminClient however target cluster use custom class custom.package.admin.TargetForwardingAdminClient:
| Code Block |
|---|
source.cluster.alias = A
target.cluster.alias = B
source.cluster.bootstrap.servers = A_localhost:9092
target.cluster.bootstrap.servers = B_localhost:9092
target.cluster.forwarding.admin.class = custom.package.admin.TargetForwardingAdminClient
// Common config for KafkaAdminClient in any ForwardingAdminClient
admin.security.protocol = SASL_SSL
admin.security.protocol=SASL_SSL
admin.sasl.mechanism=PLAIN
// Configure Source org.apache.kafka.clients.admin.ForwardingAdminClient
source.cluster.admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";
// Configure Target KafkaAdminClient in custom.package.admin.TargetForwardingAdminClient
target.cluster.admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";
// Configure Target custom ResourceManagmentRESTClient in custom.package.admin.TargetForwardingAdminClient
target.cluster.admin.resource.management.url = https://kafka.resource.manager.com
target.cluster.admin.resource.management.keystore.path = /path/keystore
target.cluster.admin.resource.management.truststore.path = /path/truststore/ca.pem |
custom.package.admin.TargetForwardingAdminClient looks like this
| Code Block | ||
|---|---|---|
| ||
package custom.package.admin;
class TargetForwardingAdminClient extends ForwardingAdminClient {
ResourceManagmentRESTClient customResourceManager;
public TargetForwardingAdminClient(Map<String, Object> configs) {
super(config)
customResourceManager = ResourceManagmentRESTClient.create(configs) //method that know how to create client.
}
@Override
public CreatePartitionsResult createPartitions(Map<String, NewPartitions> newPartitions, CreatePartitionsOptions options) {
final Map<String, KafkaFutureImpl<Void>> futures = new HashMap<>(newPartitions.size());
newPartitions.entrySet().stream().forEach(entry -> {
String topicName = entry.getKey();
int partitionCount = entry.getValue().totalCount();
customResourceManager.updateTopicPartition(topicName, partitionCount);
futures.put(topicName, new KafkaFutureImpl<>());
});
return new CreatePartitionsResult(new HashMap<>(futures));
}
// ....
} |
MirrorMaker Configuration Properties
The high-level configuration file required by the MirrorMaker driver supports the following properties:
property | default value | description |
<cluster>. | org.apache.kafka.clients.admin.ForwardingAdminClient | The fully qualified name of class that extend ForwardingAdminClient. The class must have a contractor that accept configuration (Map<String, Object> config) to configure needed clients. |
Example Configuration
| Code Block |
|---|
clusters = primary, backup
primary.bootstrap.servers = A_localhost:9092
backup.bootstrap.servers = B_localhost:9092
// Common config for KafkaAdminClient in any ForwardingAdminClient
admin.security.protocol = SASL_SSL
admin.security.protocol=SASL_SSL
admin.sasl.mechanism=PLAIN
// Configure Primary org.apache.kafka.clients.admin.ForwardingAdminClient
primary.admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";
// Configure Target KafkaAdminClient in custom.package.admin.TargetForwardingAdminClient
backup.admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";
// Configure Target custom ResourceManagmentRESTClient in custom.package.admin.TargetForwardingAdminClient
backup.admin.resource.management.url = https://kafka.resource.manager.com
backup.admin.resource.management.keystore.path = /path/keystore
backup.admin.resource.management.truststore.path = /path/truststore/ca.pem |
Compatibility, Deprecation, and Migration Plan
- When users upgrade an existing MM2 cluster they don’t need to change any of their current configuration as this proposal maintains the default behaviour for MM2.
...