Skip to main content

protobuf.decode

Decodes a field's raw data in the Confluent Protobuf format.

Description​

The processor takes raw data (bytes or a string) in the specified field and decodes it from the Confluent Protobuf wire format into structured data. It extracts the schema ID from the data, downloads the associated schema from the schema registry, resolves the schema's references and decodes the payload. The schema is cached locally after it's first compiled.

The wire format is a magic byte (0x00), a 4-byte big-endian schema ID, a message index (a zigzag-encoded varint that selects the message type within the schema), and the protobuf-encoded message payload.

If the processor encounters structured data or the data can't be decoded it returns an error.

This processor is the counterpart to the Protobuf encode capability, mirroring the schema handling of avro.decode.

If decoding fails, the processor returns a stable error code:

  • processor.protobuf_decode.schema_compile — the Protobuf schema failed to compile.
  • processor.protobuf_decode.reference_resolve — a schema reference (import) could not be resolved, or the reference graph forms a cycle.
  • processor.protobuf_decode.message_index — the Confluent message-index prefix is malformed or out of range.
  • processor.protobuf_decode.unknown_field — the payload carries a field the schema does not declare.
  • processor.protobuf_decode.wire_format — the payload is not valid Protobuf wire format for the schema.

Configuration parameters​

version: 2.2
pipelines:
  - id: example
    status: running
    connectors:
      # define source and destination ...
    processors:
      - id: example
        plugin: "protobuf.decode"
        settings:
          # The field that will be decoded.
          # For more information about the format, see [Referencing
          # fields](https://conduitdata.io/docs/using/processors/referencing-fields).
          # Type: string
          field: ".Payload.After"
          # Whether to decode the record key using its corresponding schema from
          # the schema registry.
          # Type: bool
          sdk.schema.decode.key.enabled: "true"
          # Whether to decode the record payload using its corresponding schema
          # from the schema registry.
          # Type: bool
          sdk.schema.decode.payload.enabled: "true"
          # Whether to encode the record key using its corresponding schema from
          # the schema registry.
          # Type: bool
          sdk.schema.encode.key.enabled: "true"
          # Whether to encode the record payload using its corresponding schema
          # from the schema registry.
          # Type: bool
          sdk.schema.encode.payload.enabled: "true"

Examples​

Decode a record field in Confluent Protobuf format​

This example shows the usage of the protobuf.decode processor. The processor decodes the record's.Key field using the schema that is downloaded from the schema registry and needs to exist under the subjectexample-decode. In this example we use the following schema:

syntax = "proto3";
message record {
  string myString = 1;
  int32 myInt = 2;
}

Configuration parameters​

version: 2.2
pipelines:
  - id: example
    status: running
    connectors:
      # define source and destination ...
    processors:
      - id: example
        plugin: "protobuf.decode"
        settings:
          field: ".Key"

Record difference​

Before
After
1
{
1
{
2
  "position": "dGVzdC1wb3NpdGlvbg==",
2
  "position": "dGVzdC1wb3NpdGlvbg==",
3
  "operation": "create",
3
  "operation": "create",
4
  "metadata": {
4
  "metadata": {
5
    "key1": "val1"
5
    "key1": "val1"
6
  },
6
  },
7
-
  "key": "\u0000\u0000\u0000\u0000\u0001\u0000\n\u0003bar\u0010\u0001",
7
+
  "key": {
8
+
    "myInt": 1,
9
+
    "myString": "bar"
10
+
  },
8
  "payload": {
11
  "payload": {
9
    "before": null,
12
    "before": null,
10
    "after": null
13
    "after": null
11
  }
14
  }
12
}
15
}


scarf pixel conduit-site-docs-using-processors