DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
Consider this source cluster log:
Offset | Type | isTxn | PID | Content |
0 | DATA_RECORD | true | 4001 | key=A, value=1 |
1 | DATA_RECORD | true | 4001 | key=B, value=2 |
2 | DATA_RECORD | true | 4002 | key=X, value=9 |
3 | CONTROL_MARKER | true | 4001 | COMMIT marker for PID 4001 |
4 | CONTROL_MARKER | true | 4002 | ABORT marker for PID 4002 |
5 | DATA_RECORD | false | none | key=Z, value=10 |
If replication reaches offset 4 and the source cluster fails, the destination cluster contains data records for transaction 4002 (offset 2) without the abort marker (offset 4). This creates a hanging transaction that can never be committed or aborted on the destination cluster.
...
Kafka enforces that consumer group and share group names must be unique within a single cluster. This creates a potential conflict scenario during mirroring. When such conflicts occur, the offset commit operation will fail with GroupIdNotFoundException. Users must resolve these conflicts manually by either deleting the conflicting group in the destination cluster before mirroring begins, or excluding the conflicting groups from offset synchronization. These conflicts affect only offset synchronization and do not impact data mirroring itself. The topic data continues to replicate normally, and only the automatic offset synchronization for the conflicting groups is blocked.
Diskless Topics
Given At the time of writing, the Diskless Topics KIP (KIP-1500 and other sub-KIPs) is are still under discussion, so there will be future KIPs to support it.this feature
Active-Active Writes
Active-active topology is not initially supported in Cluster Mirroring, though it could potentially be achieved through topic prefixing and removing the reliance on topic ID for mirroring. This is a candidate for a future improvement KIP.
Instead, bidirectional mirroring is supported, but only when mirroring different topics between clusters, allowing records produced to either cluster to be consumed from both. Unlike MirrorMaker 2, Cluster Mirroring does not need special cycle detection or prevention logic because the read-only enforcement inherently blocks the conditions that would create infinite replication loops.
Public Interfaces
Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.
A public interface is any change to the following:
Binary log format
The network protocol and api behavior
Any class in the public packages under clientsConfiguration, especially client configuration
org/apache/kafka/common/serialization
org/apache/kafka/common
org/apache/kafka/common/errors
org/apache/kafka/clients/producer
org/apache/kafka/clients/consumer (eventually, once stable)
Monitoring
Command line tools and arguments
- Anything else that will likely break existing users in some way when they upgrade
Proposed Changes
Command-Line
A new command-line tool kafka-mirrors.sh provides administrative operations for managing cluster mirrors.
Create a new cluster mirror configuration in the destination cluster:
| Code Block |
|---|
$ echo "bootstrap.servers=localhost:9092" >/tmp/mirror.properties
$ bin/kafka-mirror.sh --bootstrap-server :9094 --create --mirror my-mirror --mirror-config /tmp/mirror.properties |
Created mirror my-mirror
Add a topic or set of topics to an existing cluster mirror (start mirroring):
| Code Block |
|---|
$ bin/kafka-mirror.sh --bootstrap-server :9094 --add --topic my-topic --mirror my-mirror --replication-factor 2 --remote-bootstrap-server :9092 --topic-id gWrR6uDrSNSSfu_ubGndCg
Added 1 topic(s) to mirror my-mirror: [my-topic] |
List configured mirrors with additional information:
| Code Block |
|---|
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --list
MIRROR TOPICS SOURCE-BOOTSTRAP
my-mirror 2 localhost:9091
new-mirror 1 localhost:9091 |
Describe configured mirrors to check their lag compared to their source topics:
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --describe
MIRROR TOPIC PARTITION SOURCE-OFFSET DESTINATION-OFFSET LAG STATE
my-mirror bar 0 2324 2324 0 MIRRORING
my-mirror foo 0 69 66 3 MIRRORING
my-mirror foo 1 94 84 10 MIRRORING
my-mirror foo 2 94 90 4 MIRRORING
new-mirror baz 0 189 189 0 MIRRORING
new-mirror baz 1 859 859 0 MIRRORING
Remove a specific topic or set of topics from a mirror (stop mirroring / failover):
$ bin/kafka-mirror.sh --bootstrap-server :9094 --remove --topic my-topic --mirror my-mirror
Removed 1 topic(s) from mirror my-mirror: [my-topic]
Delete a mirror including its topics and configuration (stop mirroring / promotion):
TODO
Alter mirror configuration (e.g. authentication):
TODO
Throttling on the destination cluster:
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type brokers --entity-name 4 --alter --add-config mirror.replication.throttled.rate=100000000
Completed updating config for broker 4.
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type topics --entity-name my-topic --alter --add-config mirror.replication.throttled.replicas=[0:4]
Completed updating config for topic my-topic.
Throttling on the source cluster:
$ bin/kafka-configs.sh --bootstrap-server :9091 --alter --add-config 'consumer_byte_rate=1024' --entity-type clients --entity-name broker-4-fetcher-0-mirror-my-mirror
Completed updating config for client broker-4-fetcher-0-mirror-my-mirror.
Admin Client
New methods are added to the Admin interface for programmatic cluster mirror management, along with their supporting classes:
CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options);
AddTopicsToMirrorResult addTopicsToMirror(Map<String, String> topicToMirrorName, AddTopicsToMirrorOptions options);
RemoveTopicsFromMirrorResult removeTopicsFromMirror(String mirrorName, Set<String> topics, RemoveTopicsFromMirrorOptions options);
ListMirrorsResult listMirrors(ListMirrorsOptions options);
DescribeMirrorsResult describeMirrors(Collection<String> mirrorNames, DescribeMirrorsOptions options);Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.
Compatibility, Deprecation, and Migration Plan
...