Avro support #2468
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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 hidden or 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,87 @@ | ||
| /* | ||
| * Copyright 2013-present the original author or authors. | ||
| * | ||
| * Licensed 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 | ||
| * | ||
| * https://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.springframework.cloud.contract.stubrunner | ||
|
|
||
| import org.apache.avro.generic.GenericRecord | ||
| import spock.lang.Specification | ||
|
|
||
| import org.springframework.cloud.contract.verifier.messaging.avro.KafkaAvroMessageVerifierSender | ||
| import org.springframework.kafka.core.KafkaTemplate | ||
|
|
||
| class StubRunnerExecutorAvroSpec extends Specification { | ||
|
|
||
| private KafkaTemplate<String, Object> kafkaTemplate = Mock() | ||
| private KafkaAvroMessageVerifierSender sender = new KafkaAvroMessageVerifierSender(kafkaTemplate) | ||
|
|
||
| def 'should send Avro-serialized GenericRecord to Kafka for Avro contracts (bug #2404)'() { | ||
| given: | ||
| def tmpContractDir = saveTmpContract(""" | ||
| label: book_returned | ||
| input: | ||
| triggeredBy: publishBookReturned() | ||
| outputMessage: | ||
| sentTo: book.returned | ||
| headers: | ||
| X-Correlation-Id: abc-123-def | ||
|
Contributor
There was a problem hiding this comment. What headers are being sent in case of avro messages? Is there any content type?
Author
There was a problem hiding this comment. afaik, there are no headers strictly required by the kafka broker or the Avro/confluent deserializer — the schema info is embedded in the payload. This |
||
| body: | ||
| isbn: "978-1234567890" | ||
| title: "Contract Testing for Dummies" | ||
| metadata: | ||
| kafka: | ||
| avro: | ||
| schema: > | ||
| { | ||
| "type": "record", | ||
| "name": "Book", | ||
| "fields": [ | ||
| {"name": "isbn", "type": "string"}, | ||
| {"name": "title", "type": "string"} | ||
| ] | ||
| } | ||
| """) | ||
| StubRunnerExecutor executor = new StubRunnerExecutor(new AvailablePortScanner(18000, 18999), sender, []) | ||
| executor.runStubs( | ||
| new StubRunnerOptionsBuilder().build(), | ||
| new StubRepository(tmpContractDir, [], new StubRunnerOptionsBuilder().build(), null), | ||
| new StubConfiguration('avro', 'avro', 'avro', '')) | ||
| when: | ||
| executor.trigger('book_returned') | ||
| then: | ||
| 1 * kafkaTemplate.send({ | ||
| it.topic() == "book.returned" && | ||
| it.value() instanceof GenericRecord && | ||
| it.value()["schema"] != null && | ||
| it.value()["isbn"] == "978-1234567890" && | ||
| it.value()["title"] == "Contract Testing for Dummies" && | ||
| header(it, "X-Correlation-Id") == "abc-123-def" | ||
| }) | ||
| cleanup: | ||
| executor.shutdown() | ||
| tmpContractDir.deleteDir() | ||
| } | ||
|
|
||
| private File saveTmpContract(String contractYaml) { | ||
| File contractDir = File.createTempDir() | ||
| new File(contractDir, "book_returned.yml").text = contractYaml | ||
|
|
||
| contractDir | ||
| } | ||
|
|
||
| private String header(it, String key) { | ||
| new String(it.headers().lastHeader(key).value()) | ||
| } | ||
|
|
||
| } | ||
This file contains hidden or 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,22 @@ | ||
| label: book_returned | ||
| input: | ||
| triggeredBy: publishBookReturned() | ||
| outputMessage: | ||
| sentTo: book.returned | ||
| headers: | ||
| X-Correlation-Id: abc-123-def | ||
| body: | ||
| isbn: "978-1234567890" | ||
| title: "Contract Testing for Dummies" | ||
| metadata: | ||
| kafka: | ||
| avro: | ||
| schema: > | ||
| { | ||
| "type": "record", | ||
| "name": "Book", | ||
| "fields": [ | ||
| {"name": "isbn", "type": "string"}, | ||
| {"name": "title", "type": "string"} | ||
| ] | ||
| } |
This file contains hidden or 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 hidden or 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,53 @@ | ||
| /* | ||
| * Copyright 2013-present the original author or authors. | ||
| * | ||
| * Licensed 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 | ||
| * | ||
| * https://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.springframework.cloud.contract.verifier.messaging.avro; | ||
|
|
||
| /** | ||
| * Avro serialization metadata for a Kafka contract message. | ||
| * | ||
| * <p> | ||
| * Example contract YAML: <pre> | ||
| * metadata: | ||
| * kafka: | ||
| * avro: | ||
| * schema: classpath:avro/Book.avsc | ||
| * </pre> | ||
| * | ||
| * <p> | ||
| * The Schema Registry URL is configured globally via | ||
| * {@code spring.cloud.contract.avro.schema-registry-url}. | ||
| * | ||
| * @author Emanuel Trandafir | ||
| * @since 4.2.0 | ||
| */ | ||
| public class AvroMetadata { | ||
|
|
||
| /** | ||
| * Classpath or filesystem path to the Avro schema file ({@code .avsc}), e.g. | ||
| * {@code classpath:avro/Book.avsc}. May also be an inline JSON schema string. | ||
| */ | ||
| private String schema; | ||
|
|
||
| public String getSchema() { | ||
| return this.schema; | ||
| } | ||
|
|
||
| public void setSchema(String schema) { | ||
| this.schema = schema; | ||
| } | ||
|
|
||
| } |
This file contains hidden or 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,85 @@ | ||
| /* | ||
| * Copyright 2013-present the original author or authors. | ||
| * | ||
| * Licensed 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 | ||
| * | ||
| * https://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.springframework.cloud.contract.verifier.messaging.avro; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonIgnoreProperties; | ||
| import io.confluent.kafka.serializers.KafkaAvroSerializer; | ||
| import org.apache.avro.specific.SpecificRecordBase; | ||
| import org.apache.kafka.clients.producer.ProducerConfig; | ||
| import org.apache.kafka.common.serialization.StringSerializer; | ||
| import tools.jackson.databind.json.JsonMapper; | ||
|
|
||
| import org.springframework.beans.factory.ObjectProvider; | ||
| import org.springframework.beans.factory.annotation.Qualifier; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; | ||
| import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; | ||
| import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierObjectMapper; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
| import org.springframework.kafka.core.DefaultKafkaProducerFactory; | ||
| import org.springframework.kafka.core.KafkaTemplate; | ||
|
|
||
| /** | ||
| * Auto-configuration for Avro support in Spring Cloud Contract. Activates when | ||
| * {@code org.apache.avro.specific.SpecificRecordBase} is on the classpath. | ||
| * | ||
| * @author Emanuel Trandafir | ||
| * @since 4.2.0 | ||
| */ | ||
| @Configuration(proxyBeanMethods = false) | ||
| @ConditionalOnClass(name = "org.apache.avro.specific.SpecificRecordBase") | ||
| public class KafkaAvroContractVerifierConfiguration { | ||
|
|
||
| @Bean | ||
| @ConditionalOnMissingBean | ||
| ContractVerifierObjectMapper avroContractVerifierObjectMapper(ObjectProvider<JsonMapper> jsonMapper) { | ||
| JsonMapper mapper = jsonMapper.getIfAvailable(JsonMapper::new) | ||
| .rebuild() | ||
| .addMixIn(SpecificRecordBase.class, IgnoreAvroMixin.class) | ||
| .build(); | ||
| return new ContractVerifierObjectMapper(mapper); | ||
| } | ||
|
|
||
| @Bean | ||
| @ConditionalOnMissingBean(name = "avroKafkaTemplate") | ||
| KafkaTemplate<String, Object> avroKafkaTemplate(@Value("${spring.kafka.bootstrap-servers}") String bootstrapServers, | ||
|
Contributor
There was a problem hiding this comment. Why can't we reuse the users production template?
Author
There was a problem hiding this comment. I was thinking about it: this is meant to be used on the consumer side - therefore we can have a kafkaTemplate configured differently here (eg: seems unlikely, but we can have an app consuming avro and publishing json) What do you think about looking for a bean named avroKafkaTemplate first, if not, fall back to the "prod" kafkaTempalte bean, if not present either, create one. if (ctx.containsBean("avroKafkaTemplate")) {
return (KafkaTemplate<String, Object>) ctx.getBean("avroKafkaTemplate");
}
if (ctx.containsBean("kafkaTemplate")) {
return (KafkaTemplate<String, Object>) ctx.getBean("kafkaTemplate");
}
return createAvroKafkaTemplate(bootstrapServers, schemaRegistryUrl);My main point is to allow overriding this |
||
| @Value("${spring.cloud.contract.avro.schema-registry-url}") String schemaRegistryUrl) { | ||
| Map<String, Object> props = new HashMap<>(); | ||
| props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); | ||
| props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); | ||
| props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class); | ||
| props.put("schema.registry.url", schemaRegistryUrl); | ||
| return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props)); | ||
| } | ||
|
|
||
| @Bean | ||
| @ConditionalOnMissingBean | ||
| KafkaAvroMessageVerifierSender kafkaAvroMessageVerifierSender( | ||
| @Qualifier("avroKafkaTemplate") KafkaTemplate<String, Object> avroKafkaTemplate) { | ||
| return new KafkaAvroMessageVerifierSender(avroKafkaTemplate); | ||
| } | ||
|
|
||
| @JsonIgnoreProperties({ "schema", "specificData", "classSchema", "conversion" }) | ||
| interface IgnoreAvroMixin { | ||
|
Contributor
There was a problem hiding this comment. Can you provide a javadoc why we need this? |
||
|
|
||
| } | ||
|
|
||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we make this section kafka / avro agnostic? If we just set asBytes() it should work right? Another option is to verify the contentType in which case if it's not json but explicitly sth else then we just pass through? Wdyt?
I also sense that we could have some extension points here. Like check through spi (we already do it in other parts of scc) some interfaces to verify if this contract has a special way of treating payload. We could have some AvroContractPayloadProceesor that would activate when the metadata has avro and it would provide a function to how convert the payload. That way this core logic stays clean of avro but we can inject the behavior. There would have to be some priority / ordering like we do in other spis here in scc.
Wdyt?
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
by passing bytes, do you mean the 2nd approach demo-ed here?
If so - yes, it works, but it becomes very hard to peer review the contract changes and understand test failures.
In fact, if we stick to that approach, we won't need any of this - as it mainly covers workarounds needed for the 1st apporach described in the samples PR.
Are you suggesting setting a header like
contentType: application/avroand relying on it here, instead of the new custom field we added to the contract metadata?It would force users to add the contentType header to the contract even if they don't use it in prod (as it is not strictly required, more like a convention some are using).
This would work - but imo sounds a bit sketchy. wdyt?
What's your concern about using the contract meta here?
Yes, that's a great idea, actually. We have a list of processors, each of them will provide a predicate to test the contract (therefore, we see if we should apply it) , and a function for mapping/extracting the payload.
This way, we can keep using the metadata approach, but extract it somewhere else and keep this agnostic of avro.
+You mentioned potentially adding prtobuff later, which might use this extension point too.