您可以将setDeserializer方法与KafkaRecordDeserializationSchema一起使用，KafkaRecordDeserializationSchema可能如下所示：

public class KafkaUsageRecordDeserializationSchema
        implements KafkaRecordDeserializationSchema<UsageRecord> {

    private static final long serialVersionUID = 1L;

    private transient ObjectMapper objectMapper;

    @Override
    public void open(DeserializationSchema.InitializationContext context) throws Exception {
        KafkaRecordDeserializationSchema.super.open(context);
        objectMapper = JsonMapper.builder().build();
    }

    @Override
    public void deserialize(
            ConsumerRecord<byte[], byte[]> consumerRecord,
            Collector<UsageRecord> collector) throws IOException {

        collector.collect(objectMapper.readValue(consumerRecord.value(), UsageRecord.class));
    }

    @Override
    public TypeInformation<UsageRecord> getProducedType() {
        return TypeInformation.of(UsageRecord.class);
    }
}

然后可以使用ConsumerRecord访问主题和其他元数据。

我从上面的答案（大卫的）中获得灵感，并添加了以下自定义反序列化器-

KafkaSource<MyObject> source = KafkaSource.<MyObject>builder()
          .setBootstrapServers(BOOTSTRAP_SERVERS)
          .setTopics(TOPIC-1, TOPIC-2)
          .setGroupId(GROUP_ID)
          .setStartingOffsets(OffsetsInitializer.earliest())
          .setDeserializer(KafkaRecordDeserializationSchema.of(new KafkaDeserializationSchema<Event>{                                                          
    
           DeserializationSchema deserialzationSchema = ConfluentRegistryAvroDeserializationSchema.forSpecific(MyObject.class, SCHEMA_REGISTRY_URL);

           @Override
           public boolean isEndOfStream(Event nextElement) {
              return false;
           }
        
           @Override
           public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
              Event event = new Event();
              event.setTopicName(record.topic());
              event.setMyObject((MyObject) deserializationSchema.deserialize(record.value()));
              return event;
           }
        
           @Override
           public TypeInformation<String> getProducedType() {
              return TypeInformation.of(Event.class);
           }
           })).build();

Event类是MyObject类上的 Package 器，具有用于存储主题名称的附加字段。