Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -279,7 +279,7 @@ public Serializer serializer(String topic, Target type) {
@Override
public Deserializer deserializer(String topic, Target type) {
return (headers, data) -> {
var schemaId = extractSchemaIdFromMsg(data);
int schemaId = getSchemaIdFromMessageOrTopic(data, topic, type);
SchemaType format = getMessageFormatBySchemaId(schemaId);
MessageFormatter formatter = schemaRegistryFormatters.get(format);
return new DeserializeResult(
Expand All @@ -293,22 +293,30 @@ public Deserializer deserializer(String topic, Target type) {
};
}

private int getSchemaIdFromMessageOrTopic(byte[] data, String topic, Target type) {
return extractSchemaIdFromMsg(data).orElseGet(
() -> {
String subject = schemaSubject(topic, type);
return getSchemaBySubject(subject)
.map(SchemaMetadata::getId)
.orElseThrow(() -> new ValidationException(
String.format("No schema for subject '%s' found and no magic byte in avro data", subject)));
}
);
}

private SchemaType getMessageFormatBySchemaId(int schemaId) {
return getSchemaById(schemaId)
.map(ParsedSchema::schemaType)
.flatMap(SchemaType::fromString)
.orElseThrow(() -> new ValidationException(String.format("Schema for id '%d' not found ", schemaId)));
}

private int extractSchemaIdFromMsg(byte[] data) {
private Optional<Integer> extractSchemaIdFromMsg(byte[] data) {
ByteBuffer buffer = ByteBuffer.wrap(data);
if (buffer.remaining() >= SR_PAYLOAD_PREFIX_LENGTH && buffer.get() == SR_PAYLOAD_MAGIC_BYTE) {
return buffer.getInt();
return Optional.of(buffer.getInt());
}
throw new ValidationException(
String.format(
"Data doesn't contain magic byte and schema id prefix, so it can't be deserialized with %s serde",
name())
);
return Optional.empty();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,39 @@ void deserializeReturnsJsonAvroMsgJsonRepresentation() throws RestClientExceptio
.contains(Map.entry("schemaId", schemaId));
}

@Test
void deserializeReturnsJsonAvroMsgJsonRepresentationViaTopicNameOnly() throws RestClientException, IOException {
AvroSchema schema = new AvroSchema(
"{"
+ " \"type\": \"record\","
+ " \"name\": \"TestAvroRecord1\","
+ " \"fields\": ["
+ " {"
+ " \"name\": \"field1\","
+ " \"type\": \"string\""
+ " },"
+ " {"
+ " \"name\": \"field2\","
+ " \"type\": \"int\""
+ " }"
+ " ]"
+ "}"
);
String jsonValue = "{ \"field1\":\"testStr\", \"field2\": 123 }";

String topic = "test";
int schemaId = registryClient.register(topic + "-value", schema);

byte[] data = jsonToAvro(jsonValue, schema); // No magic byte no schema id registered
var result = serde.deserializer(topic, Serde.Target.VALUE).deserialize(null, data);

assertJsonsEqual(jsonValue, result.getResult());
assertThat(result.getType()).isEqualTo(DeserializeResult.Type.JSON);
assertThat(result.getAdditionalProperties())
.contains(Map.entry("type", "AVRO"))
.contains(Map.entry("schemaId", schemaId));
}

@Nested
class SerdeWithDisabledSubjectExistenceCheck {

Expand Down
Loading