-
-
Notifications
You must be signed in to change notification settings - Fork 639
/
JsonSchemaSerializer.java
70 lines (63 loc) · 2.97 KB
/
JsonSchemaSerializer.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
package org.akhq.modules.schemaregistry;
import com.fasterxml.jackson.core.JsonProcessingException;
import io.confluent.kafka.schemaregistry.ParsedSchema;
import io.confluent.kafka.schemaregistry.json.JsonSchema;
import io.confluent.kafka.serializers.json.AbstractKafkaJsonSchemaSerializer;
import lombok.extern.slf4j.Slf4j;
import org.akhq.configs.SchemaRegistryType;
import org.everit.json.schema.ValidationException;
import org.json.JSONObject;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
@Slf4j
public class JsonSchemaSerializer extends AbstractKafkaJsonSchemaSerializer<String> implements SchemaSerializer {
private final int schemaId;
private final JsonSchema jsonSchema;
private final SchemaRegistryType schemaRegistryType;
public static JsonSchemaSerializer newInstance(int schemaId, ParsedSchema parsedSchema, SchemaRegistryType schemaRegistryType) {
if (supports(parsedSchema)) {
return new JsonSchemaSerializer(schemaId, (JsonSchema) parsedSchema, schemaRegistryType);
}
String errorMsg = String.format("Schema %s has not supported schema type expected %s but found %s", parsedSchema.name(), JsonSchema.TYPE, parsedSchema.schemaType());
throw new IllegalArgumentException(errorMsg);
}
@Override
public byte[] serialize(String json) {
try {
JSONObject jsonObject = new JSONObject(json);
jsonSchema.validate(jsonObject);
} catch (JsonProcessingException e) {
String errorMsg = String.format("Provided json [%s] is not valid according to schema", json);
log.error(errorMsg);
throw new RuntimeException(errorMsg, e);
} catch (ValidationException e) {
String validationErrorMsg = String.format(
"Provided json message is not valid according to jsonSchema (id=%d): %s",
schemaId,
e.getMessage()
);
throw new IllegalArgumentException(validationErrorMsg);
}
try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
out.write(schemaRegistryType.getMagicByte());
out.write(ByteBuffer.allocate(idSize).putInt(schemaId).array());
out.write(json.getBytes(StandardCharsets.UTF_8));
byte[] bytes = out.toByteArray();
out.close();
return bytes;
} catch (IOException e) {
throw new RuntimeException(String.format("Could not serialize json [%s]", json), e);
}
}
public static boolean supports(ParsedSchema parsedSchema) {
return Objects.equals(JsonSchema.TYPE, parsedSchema.schemaType());
}
private JsonSchemaSerializer(int schemaId, JsonSchema jsonSchema, SchemaRegistryType schemaRegistryType) {
this.schemaId = schemaId;
this.jsonSchema = jsonSchema;
this.schemaRegistryType = schemaRegistryType;
}
}