Skip to content

Commit ae561bb

Browse files
eddyveddyv
authored
Merge pull request #1 from iturcino/mm2-updates
Mm2 updates (Copy of OneCricketeer#26)
2 parents 6d387cf + 2080a90 commit ae561bb

File tree

4 files changed

+102
-89
lines changed

4 files changed

+102
-89
lines changed

‎pom.xml‎

Lines changed: 31 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
<groupId>cricket.jmoore</groupId>
66
<artifactId>schema-registry-transfer-smt</artifactId>
7-
<version>0.2.1-SNAPSHOT</version>
7+
<version>0.2.1</version>
88
<name>schema-registry-transfer-smt</name>
99
<description>
1010
A Kafka Connect Transform for copying Confluent Schema Registry schemas between different registries.
@@ -66,11 +66,10 @@
6666
<maven.compiler.target>1.8</maven.compiler.target>
6767

6868
<slf4j.version>1.7.25</slf4j.version>
69-
<kafka.version>2.1.0</kafka.version>
70-
<confluent.version>5.1.0</confluent.version>
71-
<confluent.patch.version>-cp1</confluent.patch.version>
72-
<jackson.version>2.9.7</jackson.version>
73-
<jackson.asl.version>1.9.13</jackson.asl.version>
69+
<confluent.version>5.5.0</confluent.version>
70+
<jackson.version>2.10.2</jackson.version>
71+
<avro.version>1.9.2</avro.version>
72+
<jersey.bean.validation.version>2.30</jersey.bean.validation.version>
7473

7574
<spotless.version>1.19.0</spotless.version>
7675

@@ -90,14 +89,14 @@
9089
<dependency>
9190
<groupId>org.apache.kafka</groupId>
9291
<artifactId>kafka-clients</artifactId>
93-
<version>${kafka.version}${confluent.patch.version}</version>
92+
<version>${confluent.version}-ccs</version>
9493
<scope>provided</scope>
9594
</dependency>
9695

9796
<dependency>
9897
<groupId>org.apache.kafka</groupId>
9998
<artifactId>connect-transforms</artifactId>
100-
<version>${kafka.version}${confluent.patch.version}</version>
99+
<version>${confluent.version}-ccs</version>
101100
<scope>provided</scope>
102101
</dependency>
103102

@@ -122,13 +121,6 @@
122121
</exclusions>
123122
</dependency>
124123

125-
<dependency>
126-
<groupId>com.101tec</groupId>
127-
<artifactId>zkclient</artifactId>
128-
<version>0.10</version>
129-
<scope>provided</scope>
130-
</dependency>
131-
132124
<dependency>
133125
<groupId>com.fasterxml.jackson.core</groupId>
134126
<artifactId>jackson-databind</artifactId>
@@ -137,16 +129,16 @@
137129
</dependency>
138130

139131
<dependency>
140-
<groupId>org.codehaus.jackson</groupId>
141-
<artifactId>jackson-core-asl</artifactId>
142-
<version>${jackson.asl.version}</version>
132+
<groupId>com.fasterxml.jackson.core</groupId>
133+
<artifactId>jackson-core</artifactId>
134+
<version>${jackson.version}</version>
143135
<scope>provided</scope>
144136
</dependency>
145137

146138
<dependency>
147-
<groupId>org.codehaus.jackson</groupId>
148-
<artifactId>jackson-mapper-asl</artifactId>
149-
<version>${jackson.asl.version}</version>
139+
<groupId>com.fasterxml.jackson.core</groupId>
140+
<artifactId>jackson-annotations</artifactId>
141+
<version>${jackson.version}</version>
150142
<scope>provided</scope>
151143
</dependency>
152144

@@ -178,11 +170,18 @@
178170
<scope>provided</scope>
179171
</dependency>
180172

173+
<dependency>
174+
<groupId>org.glassfish.jersey.ext</groupId>
175+
<artifactId>jersey-bean-validation</artifactId>
176+
<version>${jersey.bean.validation.version}</version>
177+
<scope>provided</scope>
178+
</dependency>
179+
181180
<!-- Runtime dependencies -->
182181
<dependency>
183182
<groupId>org.apache.avro</groupId>
184183
<artifactId>avro</artifactId>
185-
<version>1.8.1</version>
184+
<version>${avro.version}</version>
186185
</dependency>
187186

188187
<dependency>
@@ -201,8 +200,16 @@
201200
<artifactId>kafka-clients</artifactId>
202201
</exclusion>
203202
<exclusion>
204-
<groupId>com.fasterxml.jackson.core</groupId>
205-
<artifactId>jackson-databind</artifactId>
203+
<groupId>io.swagger</groupId>
204+
<artifactId>swagger-annotations</artifactId>
205+
</exclusion>
206+
<exclusion>
207+
<groupId>io.swagger</groupId>
208+
<artifactId>swagger-core</artifactId>
209+
</exclusion>
210+
<exclusion>
211+
<groupId>org.glassfish.jersey.ext</groupId>
212+
<artifactId>jersey-bean-validation</artifactId>
206213
</exclusion>
207214
</exclusions>
208215
</dependency>

‎src/main/java/cricket/jmoore/kafka/connect/transforms/SchemaRegistryTransfer.java‎

Lines changed: 25 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,10 @@
2323
import org.slf4j.Logger;
2424
import org.slf4j.LoggerFactory;
2525

26+
import io.confluent.kafka.schemaregistry.ParsedSchema;
2627
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
2728
import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
28-
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
29+
import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
2930
import io.confluent.kafka.serializers.subject.TopicNameStrategy;
3031
import io.confluent.kafka.serializers.subject.strategy.SubjectNameStrategy;
3132

@@ -44,17 +45,17 @@ public class SchemaRegistryTransfer<R extends ConnectRecord<R>> implements Trans
4445

4546
public static final String SRC_PREAMBLE = "For source consumer's schema registry, ";
4647
public static final String SRC_SCHEMA_REGISTRY_CONFIG_DOC = "A list of addresses for the Schema Registry to copy from. The consumer's Schema Registry.";
47-
public static final String SRC_BASIC_AUTH_CREDENTIALS_SOURCE_CONFIG_DOC = SRC_PREAMBLE + AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE_DOC;
48-
public static final String SRC_BASIC_AUTH_CREDENTIALS_SOURCE_CONFIG_DEFAULT = AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE_DEFAULT;
49-
public static final String SRC_USER_INFO_CONFIG_DOC = SRC_PREAMBLE + AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_USER_INFO_DOC;
50-
public static final String SRC_USER_INFO_CONFIG_DEFAULT = AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_USER_INFO_DEFAULT;
48+
public static final String SRC_BASIC_AUTH_CREDENTIALS_SOURCE_CONFIG_DOC = SRC_PREAMBLE + AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE_DOC;
49+
public static final String SRC_BASIC_AUTH_CREDENTIALS_SOURCE_CONFIG_DEFAULT = AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE_DEFAULT;
50+
public static final String SRC_USER_INFO_CONFIG_DOC = SRC_PREAMBLE + AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_USER_INFO_DOC;
51+
public static final String SRC_USER_INFO_CONFIG_DEFAULT = AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_USER_INFO_DEFAULT;
5152

5253
public static final String DEST_PREAMBLE = "For target producer's schema registry, ";
5354
public static final String DEST_SCHEMA_REGISTRY_CONFIG_DOC = "A list of addresses for the Schema Registry to copy to. The producer's Schema Registry.";
54-
public static final String DEST_BASIC_AUTH_CREDENTIALS_SOURCE_CONFIG_DOC = DEST_PREAMBLE + AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE_DOC;
55-
public static final String DEST_BASIC_AUTH_CREDENTIALS_SOURCE_CONFIG_DEFAULT = AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE_DEFAULT;
56-
public static final String DEST_USER_INFO_CONFIG_DOC = DEST_PREAMBLE + AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_USER_INFO_DOC;
57-
public static final String DEST_USER_INFO_CONFIG_DEFAULT = AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_USER_INFO_DEFAULT;
55+
public static final String DEST_BASIC_AUTH_CREDENTIALS_SOURCE_CONFIG_DOC = DEST_PREAMBLE + AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE_DOC;
56+
public static final String DEST_BASIC_AUTH_CREDENTIALS_SOURCE_CONFIG_DEFAULT = AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE_DEFAULT;
57+
public static final String DEST_USER_INFO_CONFIG_DOC = DEST_PREAMBLE + AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_USER_INFO_DOC;
58+
public static final String DEST_USER_INFO_CONFIG_DEFAULT = AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_USER_INFO_DEFAULT;
5859

5960
public static final String TRANSFER_KEYS_CONFIG_DOC = "Whether or not to copy message key schemas between registries.";
6061
public static final Boolean TRANSFER_KEYS_CONFIG_DEFAULT = true;
@@ -63,7 +64,7 @@ public class SchemaRegistryTransfer<R extends ConnectRecord<R>> implements Trans
6364

6465
private CachedSchemaRegistryClient sourceSchemaRegistryClient;
6566
private CachedSchemaRegistryClient destSchemaRegistryClient;
66-
private SubjectNameStrategy<org.apache.avro.Schema> subjectNameStrategy;
67+
private SubjectNameStrategy subjectNameStrategy;
6768
private boolean transferKeys, includeHeaders;
6869

6970
// caches from the source registry to the destination registry
@@ -98,17 +99,17 @@ public void configure(Map<String, ?> props) {
9899

99100
List<String> sourceUrls = config.getList(ConfigName.SRC_SCHEMA_REGISTRY_URL);
100101
final Map<String, String> sourceProps = new HashMap<>();
101-
sourceProps.put(AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE,
102+
sourceProps.put(AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE,
102103
"SRC_" + config.getString(ConfigName.SRC_BASIC_AUTH_CREDENTIALS_SOURCE));
103-
sourceProps.put(AbstractKafkaAvroSerDeConfig.USER_INFO_CONFIG,
104+
sourceProps.put(AbstractKafkaSchemaSerDeConfig.USER_INFO_CONFIG,
104105
config.getPassword(ConfigName.SRC_USER_INFO)
105106
.value());
106107

107108
List<String> destUrls = config.getList(ConfigName.DEST_SCHEMA_REGISTRY_URL);
108109
final Map<String, String> destProps = new HashMap<>();
109-
destProps.put(AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE,
110+
destProps.put(AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE,
110111
"DEST_" + config.getString(ConfigName.DEST_BASIC_AUTH_CREDENTIALS_SOURCE));
111-
destProps.put(AbstractKafkaAvroSerDeConfig.USER_INFO_CONFIG,
112+
destProps.put(AbstractKafkaSchemaSerDeConfig.USER_INFO_CONFIG,
112113
config.getPassword(ConfigName.DEST_USER_INFO)
113114
.value());
114115

@@ -206,14 +207,14 @@ protected Optional<Integer> copySchema(ByteBuffer buffer, String topic, boolean
206207

207208
schemaAndDestId = schemaCache.get(sourceSchemaId);
208209
if (schemaAndDestId != null) {
209-
log.trace("Schema id {} has been seen before. Not registering with destination registry again.");
210+
log.trace("Schema id {} has been seen before. Not registering with destination registry again.", sourceSchemaId);
210211
} else { // cache miss
211212
log.trace("Schema id {} has not been seen before", sourceSchemaId);
212213
schemaAndDestId = new SchemaAndId();
213214
try {
214215
log.trace("Looking up schema id {} in source registry", sourceSchemaId);
215216
// Can't do getBySubjectAndId because that requires a Schema object for the strategy
216-
schemaAndDestId.schema = sourceSchemaRegistryClient.getById(sourceSchemaId);
217+
schemaAndDestId.schema = sourceSchemaRegistryClient.getSchemaById(sourceSchemaId);
217218
} catch (IOException | RestClientException e) {
218219
log.error(String.format("Unable to fetch source schema for id %d.", sourceSchemaId), e);
219220
throw new ConnectException(e);
@@ -244,25 +245,25 @@ public void close() {
244245
}
245246

246247
interface ConfigName {
247-
String SRC_SCHEMA_REGISTRY_URL = "src." + AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG;
248-
String SRC_BASIC_AUTH_CREDENTIALS_SOURCE = "src." + AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE;
249-
String SRC_USER_INFO = "src." + AbstractKafkaAvroSerDeConfig.USER_INFO_CONFIG;
250-
String DEST_SCHEMA_REGISTRY_URL = "dest." + AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG;
251-
String DEST_BASIC_AUTH_CREDENTIALS_SOURCE = "dest." + AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE;
252-
String DEST_USER_INFO = "dest." + AbstractKafkaAvroSerDeConfig.USER_INFO_CONFIG;
248+
String SRC_SCHEMA_REGISTRY_URL = "src." + AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG;
249+
String SRC_BASIC_AUTH_CREDENTIALS_SOURCE = "src." + AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE;
250+
String SRC_USER_INFO = "src." + AbstractKafkaSchemaSerDeConfig.USER_INFO_CONFIG;
251+
String DEST_SCHEMA_REGISTRY_URL = "dest." + AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG;
252+
String DEST_BASIC_AUTH_CREDENTIALS_SOURCE = "dest." + AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE;
253+
String DEST_USER_INFO = "dest." + AbstractKafkaSchemaSerDeConfig.USER_INFO_CONFIG;
253254
String SCHEMA_CAPACITY = "schema.capacity";
254255
String TRANSFER_KEYS = "transfer.message.keys";
255256
String INCLUDE_HEADERS = "include.message.headers";
256257
}
257258

258259
private static class SchemaAndId {
259260
private Integer id;
260-
private org.apache.avro.Schema schema;
261+
private ParsedSchema schema;
261262

262263
SchemaAndId() {
263264
}
264265

265-
SchemaAndId(int id, org.apache.avro.Schema schema) {
266+
SchemaAndId(int id, ParsedSchema schema) {
266267
this.id = id;
267268
this.schema = schema;
268269
}

‎src/test/java/cricket/jmoore/kafka/connect/transforms/SchemaRegistryMock.java‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
import org.slf4j.Logger;
1717
import org.slf4j.LoggerFactory;
1818

19+
import io.confluent.kafka.schemaregistry.ParsedSchema;
20+
import io.confluent.kafka.schemaregistry.avro.AvroSchema;
1921
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
2022
import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient;
2123
import io.confluent.kafka.schemaregistry.client.SchemaMetadata;
@@ -152,22 +154,22 @@ public void beforeEach(final ExtensionContext context) {
152154
.willReturn(WireMock.aResponse().withTransformers(this.getVersionHandler.getName())));
153155
this.stubFor.apply(WireMock.get(WireMock.urlPathMatching(CONFIG_PATTERN))
154156
.willReturn(WireMock.aResponse().withTransformers(this.getConfigHandler.getName())));
155-
this.stubFor.apply(WireMock.get(WireMock.urlPathMatching(SCHEMA_BY_ID_PATTERN + "\\d+"))
157+
this.stubFor.apply(WireMock.get(WireMock.urlPathMatching(SCHEMA_BY_ID_PATTERN + "\\d+/(?:fetchMaxId=false)"))
156158
.willReturn(WireMock.aResponse().withStatus(HTTP_NOT_FOUND)));
157159
}
158160

159-
public int registerSchema(final String topic, boolean isKey, final Schema schema) {
161+
public int registerSchema(final String topic, boolean isKey, final ParsedSchema schema) {
160162
return this.registerSchema(topic, isKey, schema, new TopicNameStrategy());
161163
}
162164

163-
public int registerSchema(final String topic, boolean isKey, final Schema schema, SubjectNameStrategy<Schema> strategy) {
165+
public int registerSchema(final String topic, boolean isKey, final ParsedSchema schema, SubjectNameStrategy strategy) {
164166
return this.register(strategy.subjectName(topic, isKey, schema), schema);
165167
}
166168

167-
private int register(final String subject, final Schema schema) {
169+
private int register(final String subject, final ParsedSchema schema) {
168170
try {
169171
final int id = this.schemaRegistryClient.register(subject, schema);
170-
this.stubFor.apply(WireMock.get(WireMock.urlEqualTo(SCHEMA_BY_ID_PATTERN + id))
172+
this.stubFor.apply(WireMock.get(WireMock.urlEqualTo(SCHEMA_BY_ID_PATTERN + id + "?fetchMaxId=false"))
171173
.willReturn(ResponseDefinitionBuilder.okForJson(new SchemaString(schema.toString()))));
172174
log.debug("Registered schema {}", id);
173175
return id;
@@ -242,8 +244,8 @@ public ResponseDefinition transform(final Request request, final ResponseDefinit
242244
final FileSource files, final Parameters parameters) {
243245
try {
244246
final int id = SchemaRegistryMock.this.register(getSubject(request),
245-
new Schema.Parser()
246-
.parse(RegisterSchemaRequest.fromJson(request.getBodyAsString()).getSchema()));
247+
new AvroSchema(new Schema.Parser()
248+
.parse(RegisterSchemaRequest.fromJson(request.getBodyAsString()).getSchema())));
247249
final RegisterSchemaResponse registerSchemaResponse = new RegisterSchemaResponse();
248250
registerSchemaResponse.setId(id);
249251
return ResponseDefinitionBuilder.jsonResponse(registerSchemaResponse);
@@ -279,7 +281,8 @@ private class GetVersionHandler extends SubjectsVersioHandler {
279281
@Override
280282
public ResponseDefinition transform(final Request request, final ResponseDefinition responseDefinition,
281283
final FileSource files, final Parameters parameters) {
282-
String versionStr = Iterables.get(this.urlSplitter.split(request.getUrl()), 3);
284+
String versionStrFull = Iterables.get(this.urlSplitter.split(request.getUrl()), 3);
285+
String versionStr = versionStrFull.substring(0, versionStrFull.indexOf("?"));
283286
SchemaMetadata metadata;
284287
if (versionStr.equals("latest")) {
285288
metadata = SchemaRegistryMock.this.getSubjectVersion(getSubject(request), versionStr);

0 commit comments

Comments
 (0)