Что мне делать с SourceRecords в Kafka
После создания объектов Connector и SourceTask в Kafka с Java я не могу понять, что можно сделать с SourceRecords, возвращаемым методом poll() объекта SourceTask.
Как я могу протолкнуть запись в тему, которая является параметром SourceRecocd?
Я имею в виду, что у нас есть создание SourceRecords в методе poll() следующим образом:
records.add(new SourceRecord(sourcePartition, sourceOffset, config.topicName, Schema.STRING_SCHEMA, message));
Итак, как теперь создать сообщение в теме, которая указана config.topicName выше?
Заранее спасибо.
Ответы 1
Поле topic из экземпляра SourceRecord указывает на тему, в которой будет опубликовано сообщение. В вашем случае он должен опубликовать сообщение на config.topicName.
Когда вы создаете SourceRecord, вы выбираете целевую тему. Кроме того, если вам нужно изменить его, вы можете сделать это позже через Transformations
Это означает, что если вы хотите сначала преобразовать записи, а затем опубликовать их, этого не произойдет? В этом случае создание объекта SourceRecord автоматически публикует сообщение в теме, параметры которой являются параметрами, а затем вы решаете, что делать с преобразованными записями?
Нет, в зависимости от типа соединителя поток отличается. Для SourceConnector: 1. Записи опрашиваются из внешней системы, 2. Записи преобразуются, 3. Записи преобразуются в массив байтов, для SinkConnector поток идет в обратном порядке. Вы можете прочитать больше о концепции и архитектуре: docs.confluent.io/current/connect/concepts.html
@Novemberland автоматически публикует сообщение в тему -- Да. Когда вы return records в своем коде, внутренний код платформы Connect в конечном итоге создаст Producer и отправит его в тему, которая передается в параметр SourceRecord.
Вы имеете в виду, что создание объекта SourceRecord автоматически публикует сообщение в тему?