Spring поток событий kafka с несколькими темами вывода ⇐ JAVA

[Anonymous]

У меня есть приложение Springboot, которое отправляет события Kafka в тему с определенной схемой avro. Демонстрационный код такой:

Код: Выделить всё

@Configuration
@Slf4j
public class KafkaListener {

@Bean
public Function acceptEvent() {
return messages -> {
List output = new ArrayList();
messages.forEach(msg -> {
// do work and add items to the output field
});
// some more work
return output;
};
}
}

И класс DummyClass — это класс, созданный схемой avro (файл .avsc определяется в папке ресурсов и создается плагином avro).
Итак, довольно стандартная установка. Теперь вместо отправки событий в одну тему, как я делаю это сейчас с помощью приведенного выше кода, я хочу иметь возможность отправлять события в две темы.
После поиска в Интернете я нашел пружину документация, иллюстрирующая пример использования кортежей. Поэтому я попробовал и реализовал все, кроме флюса (поскольку он мне не нужен). Новый код в итоге будет таким:

Код: Выделить всё

@Configuration
@Slf4j
public class KafkaListener {

@Bean
public Function acceptEvent() {
return messages -> {
List output1 = new ArrayList();
List output2 = new ArrayList();
messages.forEach(msg -> {
// do work and add items to the output1 and output2 fields
});
// some more work
return Tuples.of(output1,output2);
};
}
}

Однако это не работает, поскольку я получаю следующее исключение:

Код: Выделить всё

java.lang.UnsupportedOperationException: At the moment only Tuple-based function are supporting multiple arguments

Я подозреваю, что это связано с тем, что я не использую флюс. В этом ответе предполагалось, что может использоваться неправильный импорт, но я убедился, что все, что я использую, взято из правильного импорта.
Несколько другая реализация столкнулась с проблемой, аналогичной описанной в этой статье. Однако этому человеку удалось решить проблему, передав пользовательский объект, содержащий выходные аргументы в качестве полей. Я попытался сделать то же самое, но запуск приложения привел к следующему исключению:

Код: Выделить всё

Unsupported Avro type. Supported types are null, Boolean, Integer, Long, Float, Double, String, byte[] and IndexedRecord

На этом я исчерпал свои возможности. Тогда мой вопрос заключается в следующем:
Как мне добиться написания событий по нескольким темам? И если одна из моих реализаций такая, что я сделал не так? Есть ли что-нибудь еще, что можно попробовать? - Я видел KStream, в котором говорилось, что он может делать то, что я хочу, но я не могу найти хороший пример кода.

Подробнее здесь: https://stackoverflow.com/questions/791 ... put-topics