From 8b7bf44dc70fff5261b1a060340d89283dbd9599 Mon Sep 17 00:00:00 2001 From: ritikbansod2912 Date: Sun, 27 Sep 2026 16:37:54 +0530 Subject: [PATCH] feat(schema): Protobuf decode and encode for Schema Registry topics (F4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PROTOBUF subjects now work end to end alongside Avro and JSON Schema: - decode: browser, live tail and /decode render protobuf payloads as proto3 JSON with original field names; the Confluent message-index prefix (1-based varints, 0x00-terminated) is stripped, nested prefixes are reported as unsupported, and a defensive fallback parses payloads without a prefix - encode: /encode takes the subject's .proto, validates the JSON payload against the descriptor (unknown fields rejected), and emits Confluent wire bytes with the index prefix; multi-message .proto files pick the message via the wire prefix or the new optional messageName on the encode request - .proto text from the registry is compiled in-process by kafka-protobuf-provider (Confluent Community License, resolved from packages.confluent.io — new Maven repository in the pom); compiled schemas are TTL-cached like Avro schemas - tests: 5 integration tests against a mocked registry — hand-built Confluent wire fixture decodes to proto3 JSON, encode round-trips with the index prefix asserted byte-for-byte, produced messages survive produce-and-browse, unknown fields rejected, unparseable messages reported instead of thrown. Full suite 63/63. --- CHANGELOG.md | 1 + COMPATIBILITY.md | 2 +- README.md | 3 +- pom.xml | 13 ++ .../kafkawrapper/schema/ProtobufCodec.java | 183 ++++++++++++++++ .../schema/SchemaRegistryController.java | 5 +- .../schema/SchemaRegistryService.java | 34 ++- .../schema/ProtobufSchemaIntegrationTest.java | 199 ++++++++++++++++++ 8 files changed, 430 insertions(+), 10 deletions(-) create mode 100644 src/main/java/com/ritikbansod/kafkawrapper/schema/ProtobufCodec.java create mode 100644 src/test/java/com/ritikbansod/kafkawrapper/schema/ProtobufSchemaIntegrationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f211e1..c6dd078 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **Protobuf decode and encode for Schema Registry topics** (F4): PROTOBUF subjects now decode in the browser, live tail and `/decode` endpoint into proto3 JSON (original field names), and JSON payloads encode into Confluent wire bytes (message-index prefix included) on the produce path. `.proto` text from the registry is compiled in-process via `kafka-protobuf-provider` (Confluent Community License, resolved from `packages.confluent.io`); compiled schemas are TTL-cached. Multi-message `.proto` files select the message via the wire prefix or an optional `messageName` on `/encode`. Nested-message indexes are not supported yet. - **Prometheus metrics** (F6): `/actuator/prometheus` now serves `kview_*` series — per-cluster topic/partition counts, under-replicated and offline partitions, reachability (`kview_cluster_*` with a `cluster` tag), per-group consumer lag (`kview_consumer_group_lag` with `cluster`/`group` tags) and a produced-messages counter (`kview_messages_produced_total`). A 15-second in-process sampler (`KVIEW_METRICS_INTERVAL_MS`) keeps scrapes off the broker; series appear and disappear with the clusters and groups they describe. Endpoint follows the actuator auth rules (public in the default no-auth mode). - **Bulk message replay / DLQ triage** (F3): `POST /api/clusters/{clusterId}/topics/{topic}/replay` selects messages with the browser's filters and copies them to another topic — same or a different connected cluster — preserving keys, headers and original timestamps (max 1000 per call, oldest first, `dryRun` previews the selection). Exposed in the UI as a **Replay** tab in the explorer, in the CLI as `kview replay --to-topic T [...]`, and in MCP as `kview_replay_messages` (19 tools now; `confirm=true` required, `dryRun` recommended first). - **Whole-topic message search** (F2): `POST /api/clusters/{clusterId}/topics/{topic}/search` starts a bounded background scan (regex / contains / key / timestamp filters, capped at 1000 results and 2M scanned messages, max 3 concurrent jobs) with progress polling and cancellation (`GET`/`DELETE .../searches/{id}`). Exposed in the UI as a new **Search** tab in the explorer, in the CLI as `kview search` / `kview search-status`, and in MCP as `kview_search_messages` + `kview_search_status` (18 tools now). Read-only like the browser: throwaway group, never commits. diff --git a/COMPATIBILITY.md b/COMPATIBILITY.md index 13b8e77..9cc1e35 100644 --- a/COMPATIBILITY.md +++ b/COMPATIBILITY.md @@ -62,7 +62,7 @@ Create the profile in **Connections → Add cluster** (or `POST /api/clusters`). | Add-on | Status | |---|---| | **AWS MSK IAM** | Not built in (requires Amazon's proprietary `aws-msk-iam-auth` SASL mechanism). Extension point: drop the jar in, add a mechanism option mapped to `AWS_MSK_IAM` + its callback handler class. | -| **Schema Registry (Confluent/Apicurio)** | Shipped in v1.0.0 for Avro and JSON Schema over the Confluent wire format. Protobuf decoding is on the roadmap. | +| **Schema Registry (Confluent/Apicurio)** | Shipped in v1.0.0 for Avro and JSON Schema over the Confluent wire format; Protobuf decode/encode shipped as well. | | **Confluent MDS / RBAC** | Control-plane auth for Confluent tooling — unrelated to the data-plane protocol Kview uses. | | **Brokers older than 2.1** | Not supported by modern Apache clients at all (EOL). | diff --git a/README.md b/README.md index 2260e43..53b7b50 100644 --- a/README.md +++ b/README.md @@ -259,8 +259,7 @@ agent's writes are distinguishable from a human's in the audit log. ## Roadmap -- Prometheus metrics for lag and produce rate -- Protobuf decoding for Schema Registry topics (Avro and JSON Schema shipped) +- Schema registry browser: version diff, compatibility mode, upload dry-run ## Contributing diff --git a/pom.xml b/pom.xml index d01a1ac..f75305f 100644 --- a/pom.xml +++ b/pom.xml @@ -21,12 +21,25 @@ 21 + + + + confluent + https://packages.confluent.io/maven + + + org.apache.avro avro 1.11.3 + + io.confluent + kafka-protobuf-provider + 8.3.2 + com.networknt json-schema-validator diff --git a/src/main/java/com/ritikbansod/kafkawrapper/schema/ProtobufCodec.java b/src/main/java/com/ritikbansod/kafkawrapper/schema/ProtobufCodec.java new file mode 100644 index 0000000..3a1d6af --- /dev/null +++ b/src/main/java/com/ritikbansod/kafkawrapper/schema/ProtobufCodec.java @@ -0,0 +1,183 @@ +package com.ritikbansod.kafkawrapper.schema; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.util.JsonFormat; +import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema; +import io.confluent.kafka.schemaregistry.protobuf.dynamic.DynamicSchema; +import com.squareup.wire.schema.internal.parser.MessageElement; +import com.squareup.wire.schema.internal.parser.TypeElement; + +import java.util.ArrayList; +import java.util.List; + +/** + * Protobuf codec for Confluent wire-format payloads. After the 5-byte header a + * Confluent protobuf payload carries a message-index prefix — 1-based varints + * terminated by a 0x00 byte, selecting the message within the .proto file — + * followed by the protobuf message bytes. This class strips that prefix, + * parses the message against the .proto compiled from the registry, and + * converts to/from the proto3 JSON mapping (original field names preserved). + */ +public final class ProtobufCodec { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private ProtobufCodec() { } + + /** A compiled .proto: the default message name, the top-level messages in declaration order, the schema. */ + public record Compiled(String defaultName, List topLevelMessages, DynamicSchema schema) { } + + public static Compiled compile(String protoText) { + ProtobufSchema schema = new ProtobufSchema(protoText); + List topLevel = new ArrayList<>(); + for (TypeElement type : schema.rawSchema().getTypes()) { + if (type instanceof MessageElement message) { + topLevel.add(message.getName()); + } + } + return new Compiled(schema.name(), List.copyOf(topLevel), schema.toDynamicSchema()); + } + + /** + * Reads the Confluent message-index prefix and parses the message. + * Falls back to treating the whole remainder as the message body when the + * prefix parse does not yield a decodable message (non-Confluent producers). + */ + public static DynamicMessage decodeMessage(Compiled compiled, byte[] payload) throws Exception { + Indexes indexes = readIndexes(payload); + String name = resolveName(compiled, indexes.indexes()); + try { + return compiled.schema().newMessageBuilder(name) + .mergeFrom(payload, indexes.payloadOffset(), payload.length - indexes.payloadOffset()) + .build(); + } catch (Exception prefixFailure) { + if (indexes.payloadOffset() == 5 || indexes.indexes().isEmpty()) throw prefixFailure; + // some producers omit the index array entirely — retry the raw payload + return compiled.schema().newMessageBuilder(name) + .mergeFrom(payload, 5, payload.length - 5) + .build(); + } + } + + /** A message encoded for the wire: the body (index prefix + protobuf bytes) and its JSON render. */ + public record EncodedMessage(byte[] body, JsonNode json) { } + + /** Parses a JSON payload and encodes it with the Confluent index prefix for its message. */ + public static EncodedMessage encode(Compiled compiled, String messageName, JsonNode payload) throws Exception { + String name = messageName != null && !messageName.isBlank() + ? messageName : resolveDefaultName(compiled); + int index = compiled.topLevelMessages().indexOf(name); + if (index < 0) { + throw new IllegalArgumentException("messageName '" + name + + "' is not a top-level message in this .proto (found: " + compiled.topLevelMessages() + ")"); + } + DynamicMessage.Builder builder = compiled.schema().newMessageBuilder(name); + JsonFormat.parser().merge(payload.toString(), builder); + DynamicMessage message = builder.build(); + + byte[] prefix = indexPrefix(index); + byte[] messageBytes = message.toByteArray(); + byte[] body = new byte[prefix.length + messageBytes.length]; + System.arraycopy(prefix, 0, body, 0, prefix.length); + System.arraycopy(messageBytes, 0, body, prefix.length, messageBytes.length); + return new EncodedMessage(body, toJson(message)); + } + + /** Index prefix for a top-level message (Confluent writes 1-based varints, terminated by 0x00). */ + public static byte[] indexPrefix(int messageIndex) { + if (messageIndex < 0) throw new IllegalArgumentException("Negative message index"); + List bytes = new ArrayList<>(); + writeVarInt(bytes, messageIndex + 1); + bytes.add((byte) 0x00); + byte[] out = new byte[bytes.size()]; + for (int i = 0; i < bytes.size(); i++) out[i] = bytes.get(i); + return out; + } + + /** Proto3 JSON rendering with original field names (and defaults made explicit). */ + public static JsonNode toJson(DynamicMessage message) throws Exception { + String json = JsonFormat.printer() + .preservingProtoFieldNames() + .includingDefaultValueFields() + .print(message); + return MAPPER.readTree(json); + } + + static String resolveDefaultName(Compiled compiled) { + if (compiled.defaultName() != null && !compiled.defaultName().isBlank() + && compiled.topLevelMessages().contains(compiled.defaultName())) { + return compiled.defaultName(); + } + if (compiled.topLevelMessages().size() == 1) { + return compiled.topLevelMessages().get(0); + } + throw new IllegalArgumentException("The .proto defines multiple messages (" + + compiled.topLevelMessages() + ") — specify which one with messageName"); + } + + private static String resolveName(Compiled compiled, List indexes) { + if (indexes.isEmpty()) { + return resolveDefaultName(compiled); + } + if (indexes.size() > 1) { + throw new IllegalArgumentException( + "Nested protobuf message indexes are not supported (indexes " + indexes + ")"); + } + int index = indexes.get(0); + if (index < 0 || index >= compiled.topLevelMessages().size()) { + throw new IllegalArgumentException("Message index " + index + + " is outside the .proto file (" + compiled.topLevelMessages() + ")"); + } + return compiled.topLevelMessages().get(index); + } + + record Indexes(List indexes, int payloadOffset) { } + + /** Reads 1-based varints terminated by 0x00; an empty list means no index prefix was present. */ + private static Indexes readIndexes(byte[] payload) { + List indexes = new ArrayList<>(); + int pos = 5; // right after the Confluent header + while (pos < payload.length) { + int[] read = readVarInt(payload, pos); + int value = read[0]; + pos = read[1]; + if (value == 0) { + return new Indexes(indexes, pos); // terminator + } + indexes.add(value - 1); + if (indexes.size() > 64) { + throw new IllegalArgumentException("Unreasonable protobuf message index prefix (runaway varints)"); + } + } + // no terminator found — no index prefix; the payload is the message itself + return new Indexes(List.of(), 5); + } + + /** Unsigned LEB128 varint: returns {value, nextPosition}. */ + private static int[] readVarInt(byte[] bytes, int pos) { + int value = 0, shift = 0, i = pos; + while (i < bytes.length) { + int b = bytes[i] & 0xFF; + value |= (b & 0x7F) << shift; + i++; + if ((b & 0x80) == 0) return new int[]{value, i}; + shift += 7; + if (shift > 35) throw new IllegalArgumentException("Malformed protobuf index varint"); + } + throw new IllegalArgumentException("Truncated protobuf index varint"); + } + + private static void writeVarInt(List out, int value) { + while (true) { + int lower = value & 0x7F; + value >>>= 7; + if (value == 0) { + out.add((byte) lower); + return; + } + out.add((byte) (lower | 0x80)); + } + } +} diff --git a/src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryController.java b/src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryController.java index 0ff60a9..7e7eec4 100644 --- a/src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryController.java +++ b/src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryController.java @@ -108,7 +108,8 @@ public Map encode(@PathVariable String clusterId, @PathVariable throw new IllegalArgumentException("payload is required"); } SchemaRegistryService.EncodedPayload encoded = registryService.encode(clusterId, subject, - request.version(), request.payload(), key, Boolean.TRUE.equals(request.dryRun())); + request.version(), request.payload(), key, Boolean.TRUE.equals(request.dryRun()), + request.messageName()); Map body = new java.util.LinkedHashMap<>(); body.put("valueBase64", encoded.valueBase64()); body.put("schemaId", encoded.schemaId()); @@ -120,7 +121,7 @@ public Map encode(@PathVariable String clusterId, @PathVariable return body; } - public record EncodeRequestHolder(JsonNode payload, Integer version, Boolean dryRun) { } + public record EncodeRequestHolder(JsonNode payload, Integer version, Boolean dryRun, String messageName) { } /** Generates a sample JSON payload from the subject's schema. */ @GetMapping("/api/clusters/{clusterId}/registry/subjects/{subject}/sample") diff --git a/src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryService.java b/src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryService.java index 5ca6759..913a768 100644 --- a/src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryService.java +++ b/src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryService.java @@ -2,6 +2,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.protobuf.DynamicMessage; import com.networknt.schema.JsonSchema; import com.networknt.schema.JsonSchemaFactory; import com.networknt.schema.SpecVersion; @@ -154,6 +155,12 @@ public DecodedPayload decode(String clusterId, byte[] payload, String topic, boo String text = new String(payload, 5, payload.length - 5, StandardCharsets.UTF_8); result.set("decoded", AvroJsonConverter.fromJson(text)); } + case "PROTO", "PROTOBUF" -> { + var compiled = cached("proto:" + settings.get().url() + ":" + schemaId, AVRO_TTL_MS, + () -> ProtobufCodec.compile(schemaEnvelope.path("schema").asText())); + DynamicMessage message = ProtobufCodec.decodeMessage(compiled, payload); + result.set("decoded", ProtobufCodec.toJson(message)); + } default -> { result.put("note", "Decoding for '" + schemaType + "' arrives in phase 2 — showing raw payload"); result.set("decoded", AvroJsonConverter.fromJson(text(payload))); @@ -228,13 +235,19 @@ private T cached(String key, long ttlMs, java.util.function.Supplier load public record EncodedPayload(String valueBase64, int schemaId, String subject, int version, String schemaType, JsonNode normalized) { } + public EncodedPayload encode(String clusterId, String subject, Integer version, JsonNode payload, + boolean isKey, boolean dryRun) { + return encode(clusterId, subject, version, payload, isKey, dryRun, null); + } + /** - * Encodes user JSON into the registry's wire format: - * AVRO → GenericRecord binary; JSON → validated UTF-8 text. Header: 0x00 + schema id. + * Encodes a JSON payload with the subject's schema into registry wire bytes: + * AVRO → GenericRecord binary; JSON → validated UTF-8 text; PROTOBUF → + * DynamicMessage binary with the Confluent message-index prefix. * With {@code dryRun} nothing is returned for producing — the caller inspects the result. */ public EncodedPayload encode(String clusterId, String subject, Integer version, JsonNode payload, - boolean isKey, boolean dryRun) { + boolean isKey, boolean dryRun, String messageName) { SchemaRegistrySettings settings = requireSettings(clusterId); JsonNode envelope = version == null ? confluentAdapter.rawGet(settings, "/subjects/" + enc(subject) + "/versions/latest") @@ -290,8 +303,19 @@ public EncodedPayload encode(String clusterId, String subject, Integer version, body = payload.toString().getBytes(StandardCharsets.UTF_8); normalized = payload; } - case "PROTO" -> throw new IllegalArgumentException( - "Protobuf encoding arrives in phase 2b — not supported yet"); + case "PROTO", "PROTOBUF" -> { + try { + var compiled = cached("proto:" + schemaJson.hashCode(), AVRO_TTL_MS, + () -> ProtobufCodec.compile(schemaJson)); + ProtobufCodec.EncodedMessage message = ProtobufCodec.encode(compiled, messageName, payload); + body = message.body(); + normalized = message.json(); + } catch (IllegalArgumentException e) { + throw e; + } catch (Exception e) { + throw new IllegalArgumentException("Could not encode the protobuf payload: " + rootMessage(e), e); + } + } default -> throw new IllegalArgumentException("Unsupported schema type '" + schemaType + "'"); } diff --git a/src/test/java/com/ritikbansod/kafkawrapper/schema/ProtobufSchemaIntegrationTest.java b/src/test/java/com/ritikbansod/kafkawrapper/schema/ProtobufSchemaIntegrationTest.java new file mode 100644 index 0000000..56e71df --- /dev/null +++ b/src/test/java/com/ritikbansod/kafkawrapper/schema/ProtobufSchemaIntegrationTest.java @@ -0,0 +1,199 @@ +package com.ritikbansod.kafkawrapper.schema; + +import com.google.protobuf.DynamicMessage; +import com.sun.net.httpserver.HttpServer; +import com.ritikbansod.kafkawrapper.browse.MessageBrowserService; +import com.ritikbansod.kafkawrapper.connection.ClusterHandle; +import com.ritikbansod.kafkawrapper.connection.KafkaClusterManager; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +import java.io.ByteArrayOutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Phase 2b acceptance: a Confluent-compatible registry (mocked over HTTP) + * serves a PROTOBUF schema; hand-built Confluent wire bytes decode to proto3 + * JSON, and JSON payloads encode back into registry wire bytes — index prefix + * included. The wire fixture uses the Confluent shape: + * 0x00 + schema id + message-index prefix (1-based varints, 0x00-terminated) + * + protobuf message bytes. + */ +@SpringBootTest +@EmbeddedKafka(partitions = 1, topics = ProtobufSchemaIntegrationTest.TOPIC) +class ProtobufSchemaIntegrationTest { + + static final String TOPIC = "pb-orders"; + static final String SUBJECT = "pb-orders-value"; + static final int SCHEMA_ID = 7; + + static final String PROTO = """ + syntax = "proto3"; + message PbOrder { + string order_id = 1; + double amount = 2; + bool paid = 3; + } + """; + + private static HttpServer fakeRegistry; + private static String fakeRegistryUrl; + + @Autowired + SchemaRegistryService registryService; + + @Autowired + KafkaClusterManager clusterManager; + + @DynamicPropertySource + static void kafkaBootstrap(DynamicPropertyRegistry registry) { + registry.add("spring.kafka.bootstrap-servers", + () -> System.getProperty("spring.embedded.kafka.brokers")); + registry.add("kview.data-dir", () -> "target/protobuf-test-data"); + } + + @BeforeEach + void startFakeRegistryAndAttach() throws Exception { + fakeRegistry = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + fakeRegistry.createContext("/schemas/ids/" + SCHEMA_ID, exchange -> { + byte[] body = protoEnvelope().getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/vnd.schemaregistry.v1+json"); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + fakeRegistry.createContext("/subjects/" + SUBJECT + "/versions/latest", exchange -> { + byte[] body = protoEnvelope().getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/vnd.schemaregistry.v1+json"); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + fakeRegistry.createContext("/subjects", exchange -> { + byte[] body = ("[\"" + SUBJECT + "\"]").getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + fakeRegistry.start(); + fakeRegistryUrl = "http://127.0.0.1:" + fakeRegistry.getAddress().getPort(); + + registryService.save(KafkaClusterManager.DEFAULT_CLUSTER_ID, + SchemaRegistrySettings.confluent(fakeRegistryUrl)); + } + + @AfterEach + void detachAndStop() { + registryService.delete(KafkaClusterManager.DEFAULT_CLUSTER_ID); + if (fakeRegistry != null) fakeRegistry.stop(0); + } + + private static String protoEnvelope() { + return "{\"schemaType\":\"PROTOBUF\",\"id\":" + SCHEMA_ID + + ",\"schema\":" + com.fasterxml.jackson.databind.node.TextNode.valueOf(PROTO).toString() + "}"; + } + + private byte[] header(int schemaId) { + return new byte[]{0x00, + (byte) ((schemaId >>> 24) & 0xFF), (byte) ((schemaId >>> 16) & 0xFF), + (byte) ((schemaId >>> 8) & 0xFF), (byte) (schemaId & 0xFF)}; + } + + /** Builds the protobuf message bytes through the same codec the service uses. */ + private byte[] protobufMessage(String orderId, double amount, boolean paid) throws Exception { + var compiled = ProtobufCodec.compile(PROTO); + var descriptor = compiled.schema().getMessageDescriptor("PbOrder"); + return DynamicMessage.newBuilder(descriptor) + .setField(descriptor.findFieldByName("order_id"), orderId) + .setField(descriptor.findFieldByName("amount"), amount) + .setField(descriptor.findFieldByName("paid"), paid) + .build() + .toByteArray(); + } + + @Test + void confluentWireBytesDecodeToProto3Json() throws Exception { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + out.write(header(SCHEMA_ID)); + out.write(new byte[]{0x01, 0x00}); // message-index prefix: top-level message 0 + out.write(protobufMessage("ORD-42", 91.5, true)); + + var decoded = registryService.decode(KafkaClusterManager.DEFAULT_CLUSTER_ID, + out.toByteArray(), TOPIC, false); + + assertThat(decoded.wireFormat()).isEqualTo("CONFLUENT"); + assertThat(decoded.error()).isNull(); + assertThat(decoded.decoded().get("order_id").asText()).isEqualTo("ORD-42"); + assertThat(decoded.decoded().get("amount").asDouble()).isEqualTo(91.5); + assertThat(decoded.decoded().get("paid").asBoolean()).isTrue(); + assertThat(decoded.subject()).isEqualTo(SUBJECT); + } + + @Test + void encodeProducesConfluentWireBytesThatDecodeBack() throws Exception { + var payload = new com.fasterxml.jackson.databind.ObjectMapper() + .readTree("{\"order_id\":\"ENC-1\",\"amount\":55.25,\"paid\":true}"); + + var encoded = registryService.encode(KafkaClusterManager.DEFAULT_CLUSTER_ID, SUBJECT, + null, payload, false, false); + + assertThat(encoded.schemaType()).isEqualTo("PROTOBUF"); + byte[] wire = java.util.Base64.getDecoder().decode(encoded.valueBase64()); + assertThat(wire[0]).isEqualTo((byte) 0x00); + assertThat(wire[1]).isEqualTo((byte) 0); assertThat(wire[2]).isEqualTo((byte) 0); + assertThat(wire[3]).isEqualTo((byte) 0); assertThat(wire[4]).isEqualTo((byte) SCHEMA_ID); + assertThat(wire[5]).isEqualTo((byte) 0x01); // index prefix: top-level message 0, 1-based + assertThat(wire[6]).isEqualTo((byte) 0x00); // terminator + + // round trip through the decoder + var decoded = registryService.decode(KafkaClusterManager.DEFAULT_CLUSTER_ID, wire, TOPIC, false); + assertThat(decoded.decoded().get("order_id").asText()).isEqualTo("ENC-1"); + assertThat(decoded.decoded().get("amount").asDouble()).isEqualTo(55.25); + } + + @Test + void encodedMessagesSurviveProduceAndBrowse() throws Exception { + var payload = new com.fasterxml.jackson.databind.ObjectMapper() + .readTree("{\"order_id\":\"PB-1\",\"amount\":10.5,\"paid\":false}"); + var encoded = registryService.encode(KafkaClusterManager.DEFAULT_CLUSTER_ID, SUBJECT, + null, payload, false, false); + + ClusterHandle handle = clusterManager.get(KafkaClusterManager.DEFAULT_CLUSTER_ID); + handle.template().send(new org.apache.kafka.clients.producer.ProducerRecord<>(TOPIC, + "pb-1".getBytes(), java.util.Base64.getDecoder().decode(encoded.valueBase64()))).get(); + + var result = new MessageBrowserService(registryService).browse(handle, TOPIC, + new MessageBrowserService.BrowseRequest(null, "earliest", null, null, 10, 10_000L, null, null)); + assertThat(result.messages()).hasSize(1); + assertThat(result.messages().get(0).schema().decoded().get("order_id").asText()).isEqualTo("PB-1"); + assertThat(result.messages().get(0).schema().schemaId()).isEqualTo((long) SCHEMA_ID); + } + + @Test + void unknownJsonFieldsAreRejected() throws Exception { + var payload = new com.fasterxml.jackson.databind.ObjectMapper() + .readTree("{\"totally_unknown\":1}"); + assertThatThrownBy(() -> registryService.encode(KafkaClusterManager.DEFAULT_CLUSTER_ID, SUBJECT, + null, payload, false, false)) + .isInstanceOf(Exception.class) + .hasMessageContaining("totally_unknown"); + } + + @Test + void unparseableMessageIsReportedNotThrown() { + byte[] wire = new byte[]{0x00, 0x00, 0x00, 0x00, 0x07, 0x01, 0x00, 0x00, 0x00}; + var result = registryService.decode(KafkaClusterManager.DEFAULT_CLUSTER_ID, wire, TOPIC, false); + assertThat(result.decoded()).isNull(); + assertThat(result.error()).isNotNull(); + } +}