feat(schema): Protobuf decode and encode for Schema Registry topics

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 <topic> --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.

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). |
 

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
 

pom.xml

@@ -21,12 +21,25 @@
     <java.version>21</java.version>
   </properties>
 
+  <repositories>
+    <!-- Confluent's schema-registry protobuf provider (runtime .proto parsing) -->
+    <repository>
+      <id>confluent</id>
+      <url>https://packages.confluent.io/maven</url>
+    </repository>
+  </repositories>
+
   <dependencies>
     <dependency>
       <groupId>org.apache.avro</groupId>
       <artifactId>avro</artifactId>
       <version>1.11.3</version>
     </dependency>
+    <dependency>
+      <groupId>io.confluent</groupId>
+      <artifactId>kafka-protobuf-provider</artifactId>
+      <version>8.3.2</version>
+    </dependency>
     <dependency>
       <groupId>com.networknt</groupId>
       <artifactId>json-schema-validator</artifactId>

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<String> topLevelMessages, DynamicSchema schema) { }
+
+    public static Compiled compile(String protoText) {
+        ProtobufSchema schema = new ProtobufSchema(protoText);
+        List<String> 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<Byte> 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<Integer> 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<Integer> 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<Integer> 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<Byte> out, int value) {
+        while (true) {
+            int lower = value & 0x7F;
+            value >>>= 7;
+            if (value == 0) {
+                out.add((byte) lower);
+                return;
+            }
+            out.add((byte) (lower | 0x80));
+        }
+    }
+}

src/main/java/com/ritikbansod/kafkawrapper/schema/SchemaRegistryController.java

@@ -108,7 +108,8 @@ public Map<String, Object> 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<String, Object> body = new java.util.LinkedHashMap<>();
         body.put("valueBase64", encoded.valueBase64());
         body.put("schemaId", encoded.schemaId());
@@ -120,7 +121,7 @@ public Map<String, Object> 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")

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> T cached(String key, long ttlMs, java.util.function.Supplier<T> 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 + "'");
         }