root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-14745: Cache the ReplicationPolicy instance in MirrorConnectorConfig (#13328) 

Reviewers: Chris Egerton <fearthecellos@gmail.com>
This commit is contained in:
Mickael Maison 2023-03-03 12:14:17 +01:00 committed by GitHub
parent 517b5d2b09
commit 71fa008b45
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 11 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorConnectorConfig.java
View File

@ -108,8 +108,11 @@ public abstract class MirrorConnectorConfig extends AbstractConfig {
public static final String OFFSET_SYNCS_TOPIC_LOCATION_DEFAULT = SOURCE_CLUSTER_ALIAS_DEFAULT; public static final String OFFSET_SYNCS_TOPIC_LOCATION_DEFAULT = SOURCE_CLUSTER_ALIAS_DEFAULT;
public static final String OFFSET_SYNCS_TOPIC_LOCATION_DOC = "The location (source/target) of the offset-syncs topic."; public static final String OFFSET_SYNCS_TOPIC_LOCATION_DOC = "The location (source/target) of the offset-syncs topic.";
private final ReplicationPolicy replicationPolicy;
protected MirrorConnectorConfig(ConfigDef configDef, Map<String, String> props) { protected MirrorConnectorConfig(ConfigDef configDef, Map<String, String> props) {
super(configDef, props, true); super(configDef, props, true);
replicationPolicy = getConfiguredInstance(REPLICATION_POLICY_CLASS, ReplicationPolicy.class);
} }
String connectorName() { String connectorName() {
@ -133,7 +136,7 @@ public abstract class MirrorConnectorConfig extends AbstractConfig {
} }
ReplicationPolicy replicationPolicy() { ReplicationPolicy replicationPolicy() {
return getConfiguredInstance(REPLICATION_POLICY_CLASS, ReplicationPolicy.class); return replicationPolicy;
} }
Map<String, Object> sourceProducerConfig() { Map<String, Object> sourceProducerConfig() {

4
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConfig.java
View File

@ -171,10 +171,6 @@ public class MirrorSourceConfig extends MirrorConnectorConfig {
} }
} }
ReplicationPolicy replicationPolicy() {
return getConfiguredInstance(REPLICATION_POLICY_CLASS, ReplicationPolicy.class);
}
int replicationFactor() { int replicationFactor() {
return getInt(REPLICATION_FACTOR); return getInt(REPLICATION_FACTOR);
} }

7
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorConnectorConfigTest.java
View File

@ -29,6 +29,7 @@ import static org.apache.kafka.connect.mirror.TestUtils.makeProps;
import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertSame;
public class MirrorConnectorConfigTest { public class MirrorConnectorConfigTest {
@ -183,4 +184,10 @@ public class MirrorConnectorConfigTest {
assertEquals(2, config.metricsReporters().size()); assertEquals(2, config.metricsReporters().size());
} }
@Test
public void testReplicationPolicy() {
MirrorConnectorConfig config = new TestMirrorConnectorConfig(makeProps());
assertSame(config.replicationPolicy(), config.replicationPolicy());
}
} }