Я начинающий с Кафкой.Я использую Kafka Consumer в моей автоматизации, я могу получать сообщения, но это сериализованные данные (в байтах).Я хотел бы десериализовать его и получить сообщение в формате pretty / json.Как я должен сделать это в рубине, любые ведет?Также я использую следующий код, чтобы остановить чтение от потребителя, но моя автоматизация не останавливается и продолжает работать:
consumer.subscribe(topic)
trap('TERM') {consumer.stop}
Спасибо, RV