-
Notifications
You must be signed in to change notification settings - Fork 14.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
KAFKA-15372: Reconfigure dedicated MM2 connectors after leadership ch…
…ange (#14293) Signed-off-by: Greg Harris <[email protected]> Reviewers: Chris Egerton <[email protected]>
- Loading branch information
1 parent
68389c2
commit 3f7eada
Showing
4 changed files
with
160 additions
and
27 deletions.
There are no files selected for viewing
90 changes: 90 additions & 0 deletions
90
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorHerder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.kafka.connect.mirror; | ||
|
|
||
| import org.apache.kafka.common.utils.Time; | ||
| import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy; | ||
| import org.apache.kafka.connect.runtime.Worker; | ||
| import org.apache.kafka.connect.runtime.distributed.DistributedConfig; | ||
| import org.apache.kafka.connect.runtime.distributed.DistributedHerder; | ||
| import org.apache.kafka.connect.runtime.distributed.NotLeaderException; | ||
| import org.apache.kafka.connect.runtime.rest.RestClient; | ||
| import org.apache.kafka.connect.storage.ConfigBackingStore; | ||
| import org.apache.kafka.connect.storage.StatusBackingStore; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| import static org.apache.kafka.connect.mirror.MirrorMaker.CONNECTOR_CLASSES; | ||
|
|
||
| public class MirrorHerder extends DistributedHerder { | ||
|
|
||
| private static final Logger log = LoggerFactory.getLogger(MirrorHerder.class); | ||
|
|
||
| private final MirrorMakerConfig config; | ||
| private final SourceAndTarget sourceAndTarget; | ||
| private boolean wasLeader; | ||
|
|
||
| public MirrorHerder(MirrorMakerConfig mirrorConfig, SourceAndTarget sourceAndTarget, DistributedConfig config, Time time, Worker worker, String kafkaClusterId, StatusBackingStore statusBackingStore, ConfigBackingStore configBackingStore, String restUrl, RestClient restClient, ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy, List<String> restNamespace, AutoCloseable... uponShutdown) { | ||
| super(config, time, worker, kafkaClusterId, statusBackingStore, configBackingStore, restUrl, restClient, connectorClientConfigOverridePolicy, restNamespace, uponShutdown); | ||
| this.config = mirrorConfig; | ||
| this.sourceAndTarget = sourceAndTarget; | ||
| } | ||
|
|
||
| @Override | ||
| protected void rebalanceSuccess() { | ||
| if (isLeader()) { | ||
| if (!wasLeader) { | ||
| log.info("This node {} is now a leader for {}. Configuring connectors...", this, sourceAndTarget); | ||
| configureConnectors(); | ||
| } | ||
| wasLeader = true; | ||
| } else { | ||
| wasLeader = false; | ||
| } | ||
| } | ||
|
|
||
| private void configureConnectors() { | ||
| CONNECTOR_CLASSES.forEach(this::maybeConfigureConnector); | ||
| } | ||
|
|
||
| private void maybeConfigureConnector(Class<?> connectorClass) { | ||
| Map<String, String> desiredConfig = config.connectorBaseConfig(sourceAndTarget, connectorClass); | ||
| Map<String, String> actualConfig = configState.connectorConfig(connectorClass.getSimpleName()); | ||
| if (actualConfig == null || !actualConfig.equals(desiredConfig)) { | ||
| configureConnector(connectorClass.getSimpleName(), desiredConfig); | ||
| } else { | ||
| log.info("This node is a leader for {} and configuration for {} is already up to date.", sourceAndTarget, connectorClass.getSimpleName()); | ||
| } | ||
| } | ||
|
|
||
| private void configureConnector(String connectorName, Map<String, String> connectorProps) { | ||
| putConnectorConfig(connectorName, connectorProps, true, (e, x) -> { | ||
| if (e == null) { | ||
| log.info("{} connector configured for {}.", connectorName, sourceAndTarget); | ||
| } else if (e instanceof NotLeaderException) { | ||
| // No way to determine if the herder is a leader or not beforehand. | ||
| log.info("This node lost leadership for {} while trying to update the connector configuration for {}. Using existing connector configuration.", connectorName, sourceAndTarget); | ||
| } else { | ||
| log.error("Failed to configure {} connector for {}", connectorName, sourceAndTarget, e); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters