Need help for Flink-Connector-Fulsar

[ThinhLe30]

I'm using Flink to stream processing event from Apache Pulsar (using Avro Schema) to Cassandra database.
There's the problem:

• When I init new PulsarSoure and read the data stream from Pulsar, I'm only can touch the Message payload or get the GenericReocord, which only touches the record payload
• But in the Pulsar Producer to produce a message, I also put the data on Properties field, like below:/

msgID, err := p.producer.Send(ctx, &pulsar.ProducerMessage{
		Key:        aggregateID.String(),
		Payload:    value,
		EventTime:  event.Timestamp(),
		Properties: properties,
	})

And the Flink job:

the schema Deserializer:

public class BlockDeserializationSchema implements DeserializationSchema<Block> {
    private static final Logger LOGGER = LoggerFactory.getLogger(BlockDeserializationSchema.class);
    private static final long serialVersionUID = 1L;

    private static CachedSchemaRegistryClient client = EnvironmentUtils.initSchemaRegistry();

    public BlockDeserializationSchema() {
    }

    @Override
    public Block deserialize(byte[] bytes) throws IOException {
        /* TODO:
        - Decode 5 first bytes to get schema registry id
        - Fetch avro schema def form schema registry
        - Decode payload
        * */
//        String schemaID = new String(bytes, 0, 5, StandardCharsets.UTF_8);

        Block block = null;
        try {
            ParsedSchema schema = client.getSchemaById(6);
            Schema avroSchema = new Schema.Parser().parse(schema.canonicalString());
            DatumReader<Block> datumReader = new GenericDatumReader<>(avroSchema);
            Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
            try {
                block = datumReader.read(null, decoder);
                System.out.println(block.getId());
            } catch (IOException e) {
                e.printStackTrace();
                return null;
            }

        } catch (RestClientException e) {
            throw new RuntimeException(e);
        }

        return block;
    }

    @Override
    public boolean isEndOfStream(Block block) {
        return false;
    }

It's error, so I wanna ask:
How can I get POJO class from byte[] bytes and get the properties in the Flink Job, if you have any example, I'm very thankful

[ThinhLe30]

Update:
Use PulsarDeserializationSchema<T> is solve the prob, thanks <3. But deserialize the byte[] to pojo is still open