Пакетные записи с шаблоном kafka

Question

У меня есть приложение на стороне сервера java, которое прослушивает tcp-соединение, читает сообщения, обрабатывает и публикует тему kafka.Мы используем транзакционный шаблон kafka с kafka tm, чтобы достичь только семантики. Это прекрасно работает, когда мы фиксируем запись по записи. Мы хотим зафиксировать пакет записей либо на основе события (проверьте, есть ли в сообщении определенный набор параметров, затем зафиксировать) или на основе временного интервала.Это то, что возможно с шаблоном kafka. Мы используем пружину kafka 2.1

Answer 1

Одно решение:

Вы можете запустить KafkaTemplate.executeInTransaction() в другом потоке и передать запросы от прослушивателя TCP через BlockingQueue или подобное, с очередью poll(), измененной в рамках обратного вызова (возможно, с таймаутом ).

Тогда, когда применяется правильное условие; выйдите из функции обратного вызова, чтобы зафиксировать транзакцию, затем запустите другую.