Могу ли я использовать один и тот же файл state.dir для нескольких экземпляров в потоках Kafka?

Question

Мой вариант использования - получить последнее значение для любой ключ в теме Кафки.Для этого я в настоящее время использую GlobalStateStore.

Проблема в том, что, если у меня разные application.id для каждого экземпляра, создается другой каталог состояния.

То, что я хотел бы, этоповторно использовать существующую директорию хранилища состояний во всех экземплярах, чтобы избежать дублирования данных.Это потому, что все мои экземпляры приложений работают на одном компьютере, поэтому нет смысла иметь разные каталоги состояний.

Я сделал следующее:

Используйте тот же application.id и создали 2 экземпляра потока. Эти 2 экземпляра указывают на один и тот же файл state.dir, но экземпляры хранилища состояний разные (объекты java разные) Statestore1 и Statestore2

, а затем в качестве теста я сделал следующее:

while(true)
{
 new Thread(()-> stateStore1.get(key)).start();
 new Thread(()-> stateStore2.get(key)).start();
}

и затем я начал производить значения в теме ...

Я заметил, что всеЭкземпляры хранилища состояний (Statestore1 и Statestore2) обновляют одно и то же хранилище состояний (т.е. тот же State.dir) (так как это код)

Я не столкнулся с какими-либо проблемами в тестах, выполняющих это.Но ... я предполагаю, что блокировка записи должна быть получена до записи данных в файл state.dir.Теперь, если по какой-то причине, Statestore1 не снял блокировку, то StatesStore2 будет продолжать ждать?

Мой подход выше безопасен или есть другие способы?

Обновление:

Вариант использования состоит в том, что есть некоторые общие данные, которыми я хотел бы поделиться в разных процессах (разных экземплярах JVM).Поскольку данные являются общими, нет необходимости иметь глобальное хранилище для каждого процесса (поскольку оно будет избыточным).

Все процессы должны иметь возможность чтения и записи в это хранилище, поскольку каждый процесс не зависит отдруг с другом.Поэтому, если один процесс завершается, другой может выполнять свою работу (сохранить последнее значение в глобальном хранилище и получить его при необходимости).

Answer 1

Прежде всего, state_dir принимает параметр каталога, в котором будет создана папка со значением application.id.

Так, например, если state_dir равно C:\tmp и application.id - это test , тогда папка будет C:\tmp\test

Итак, если для state_dir задано значение C:\tmp\test, а не C:\tmp сто же самое application.id или даже другое application.id, новый каталог будет создан в C:\tmp\test\ как C:\tmp\test\test или C:\tmp\test\some_other_application_id.

Они не конфликтуют друг с другом.Все еще будет несколько глобальных хранилищ.

Однако, если state_dir задано равным C:\tmp для обоих KafkaStreams экземпляров, то возникнет исключение, сообщающее

Не удалось заблокировать каталог глобального состояния.Это может произойти, если несколько экземпляров KafkaStreams выполняются на одном хосте, используя один и тот же каталог состояний

В случае использования общего места для всех экземпляров лучше использовать базу данных какпереход на Kafka (с использованием API подключения Kafka или записи из потоков в базу данных вручную).

Если сценарий использования не требует сложных SQL-подобных запросов, то вы можете иметь свой собственный RocksDB, которыйбудет выполнять запись в один и тот же каталог для всех экземпляров, работающих на одном компьютере.

Кажется, что Kafka не поддерживает такой вариант использования "из коробки".

Answer 2

Если вы запускаете все на одном компьютере и хотите использовать глобальное хранилище, вы не должны использовать несколько экземпляров, а несколько потоков в одном экземпляре.

Глобальные хранилища предназначены для репликации для каждого экземпляра.и это именно то, чего вы хотите избежать.