我正在尝试连接kafak主题，它使用avro来触发流。我试着用下面的代码：

schema_registry_client = CachedSchemaRegistryClient(url='schema_registry_url')
serializer = MessageSerializer(schema_registry_client)

def decoder(s):
    decoded_message = serializer .decode_message(s)
    return decoded_message

kvs = KafkaUtils.createDirectStream(ssc, topic, {'bootstrap.servers': brokers}, valueDecoder=decoder)

但是，当我运行此命令时出现以下错误：
“kafka dstream”对象没有属性“\u get\u object\u id”
我们怎样才能克服这个问题？