Hi I am having trouble decoding protobuf values in pyflink received from the kafka data streaming connector: I am trying to deserialise a evaluation type:

// Representation of a Model Evaluation that was made:
// "This model produced this output when given this input".
message Evaluation {
    string id = 1;
    Model model = 2;
    ModelIO io = 3;
    CorrelationData correlation_data = 4;
    google.protobuf.Timestamp timestamp = 5;
}

message Model {
    string name = 1;
    string version = 2;
}

message ModelIO {
    google.protobuf.Struct input = 1;
    google.protobuf.Struct output = 2;
}

message CorrelationData {
    google.protobuf.Struct data = 1;
}

Evaluations store timestamps as

google.protobuf.Timestamp

typed values. Evaluations are serialised as protobuf byte arrays and are written to a evaluations kafka topic as kafka messages:

{
key: evaluation.id
msg: evaluation
}

Consumers are able to consume from topics. When they do they receive the kafka message object, it is serialised as a byte string/array/vector of the structure described above. When Flink java process consumes these message it will encode the

msg

object as a

UTF-8

string and pass the encoded string to python so that PyFlink can consume the message. We are then able to deserialise the object:

def deserialise(value):
        bytes = value.encode('utf-8')
        evaluation = Evaluation.FromString(bytes)
        return evaluation

However an error is raised:

*** google.protobuf.message.DecodeError: Error parsing message with type 'evaluation.Evaluation'

This is because the timestamp field is not correctly handled. The byte string looks like this:

b'\n\npxgqdrnbtj\x12\x06\n\x01c\x12\x016\x1a3\n\x17\n\x15\n\x05input\x12\x0c\x1a\npoietemwix\x12\x18\n\x16\n\x06output\x12\x0c\x1a\nhbzualvwbz"\x17\n\x15\n\x13\n\x03ref\x12\x0c\x1a\nnoqqltgkfq*\x0c\x08\xef\xbf\xbd\xd1\xa7\xef\xbf\xbd\x06\x10\xef\xbf\xbd\xef\xbf\xbd\xef\xbf\xbd\xef\xbf\xbd\x03'

The last 26 bytes (everything from and including the

*

) encode the timestamp field. If we try to deserialise the object whilst omitting the timestamp field, no error is raised:

python
(pdb) Evaluation.FromString(bytes[:-26])
id: "pxgqdrnbtj"
model {
  name: "c"
  version: "6"
}
io {
  input {
    fields {
      key: "input"
      value {
        string_value: "poietemwix"
      }
    }
  }
  output {
    fields {
      key: "output"
      value {
        string_value: "hbzualvwbz"
      }
    }
  }
}
correlation_data {
  data {
    fields {
      key: "ref"
      value {
        string_value: "noqqltgkfq"
      }
    }
  }
}

I create a second script which uses

confluent-kafka.Consumer

and attempt to deserialised the

msg

field of a kafka message.

consumer = Consumer({
        "bootstrap.servers": BOOTSTRAP_SERVER,
        "client.id": "sample-producer",
        "group.id": "consumer",
    })

    consumer.subscribe([TOPIC])

    while True:
        try:
            msg = consumer.poll(1.0)
            if not msg:
                continue
            print(msg.value())
            breakpoint()
        except KeyboardInterrupt:
            break

We can copy the byte string produced by this script into the flink program debug context:

python
(pdb) bs = b'\n\nadtiegqpdd\x12\x06\n\x01c\x12\x015\x1a3\n\x17\n\x15\n\x05input\x12\x0c\x1a\noscqcjntwt\x12\x18\n\x16\n\x06output\x12\x0c\x1a\ngwwkwcizol"\x17\n\x15\n\x13\n\x03ref\x12\x0c\x1a\ncsicpsesyc*\x0c\x08\xd5\xc6\xab\xa4\x06\x10\xe8\x9f\xa9\xba\x03'
(pdb) Evaluation.FromString(bs)
id: "adtiegqpdd"
model {
  name: "c"
  version: "5"
}
io {
  input {
    fields {
      key: "input"
      value {
        string_value: "oscqcjntwt"
      }
    }
  }
  output {
    fields {
      key: "output"
      value {
        string_value: "gwwkwcizol"
      }
    }
  }
}
correlation_data {
  data {
    fields {
      key: "ref"
      value {
        string_value: "csicpsesyc"
      }
    }
  }
}
timestamp {
  seconds: 1686823765
  nanos: 927617000
}

This shows the protobuf timestamp type is available on the filesystem. Therefore I believe the issue is related to how PyFlink is reading kafka messages. Pyflinks kafka consumer is a java class which is called from a python process. - The python representation configures initialises a java object via

java gateway

, by configuring its charset to UTF-8. - The Java backend simply serialises the value fields and returns it to caller (source). I believe there is an issue with the encoding but I am currently unsure. Has anyone has any similar issues when handling protobuf messages from kafka?

@Dominic Lindsay Are you able to check if the bytes received in Java and the bytes deserialized in Python are the same? It could help to verify if it’s the encoding problem.

@Dian Fu I could set these an envrionment vars, however neither SimpleStringSchema or FlinkKafkaConsumer log the received bytes.

Yes, I guess we have to debug it …

So then I would need to recompile flink?

You need download the source code and import it to an IDE for debugging, however, not necessary to compile it.