Кафка присоединиться к хранилищу

Question

• 1000 1006 *, (выглядит как RocksDB) и постоянно растет . Кроме того, хранилище состояний в кластере Kafka постоянно растет .

Итак, я изменил реализацию соединения потоков на:

...
private final long retentionHours = Duration.ofDays(3);
    ...
    var joinWindow = JoinWindows.of(Duration.ofMinutes(retentionHours))
                                .grace(Duration.ofMillis(0));
    var joinStores = StreamJoined.with(Serdes.String(), aggregatorSerde, aggregatorSerde)
                                 .withStoreName("STORE-1")
                                 .withName("STORE-2")
                                 .withThisStoreSupplier(createStoreSupplier("MEM-STORE-1"))
                                 .withOtherStoreSupplier(createStoreSupplier("MEM-STORE-2"));
    stream1.join(stream2, streamJoiner(), joinWindow, joinStores);

...
private WindowBytesStoreSupplier createStoreSupplier(String storeName) {
    var window = Duration.ofMinutes(retentionHours * 2)
                         .toMillis();
    return new InMemoryWindowBytesStoreSupplier(storeName, window, window, true);
}

Теперь нет папки состояний: / tmp / kafka-streams .

Означает ли это, что InMemoryWindowBytesStoreSupplier вообще не использует диск? Если да, то как это работает?

Кроме того, я все еще вижу, что хранилище состояний в кластере Kafka постоянно растет .

Answer 1

Означает ли это, что InMemoryWindowBytesStoreSupplier вообще не использует диск? Если да, то как это работает?

IIR C, InMemoryWindowBytesStore вообще не использует диск.

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

Кроме того, я все еще вижу, что хранилище состояний в кластере Kafka постоянно растет.

Однако InMemoryWindowBytesStore по-прежнему отказоустойчив. Это часто сбивает с толку новых разработчиков Kafka Streams, потому что в большинстве программ "в памяти" всегда подразумевает, что "данные теряются, если что-то происходит. ". Однако это не относится к Kafka Streams. Хранилище состояний всегда постоянно «резервируется» в его журнал изменений Kafka c, независимо от того, используете ли вы хранилище состояний по умолчанию (с RocksDB) или хранилище состояний в памяти. Это объясняет, почему вы видите данные состояния (журнала изменений) в памяти в кластере Kafka . Данные не должны расти вечно, кстати, поскольку темы журнала изменений сжаты чтобы предотвратить именно этот сценарий.

Примечание. Однако при использовании хранилища в памяти экземпляры вашего приложения могут работать вне памяти (OOM), и, следовательно, cra sh. Хотя ваши данные о состоянии никогда не будут потеряны, как объяснено выше, ваше приложение не будет работать из-за ошибки OOM sh / оно будет работать только частично (некоторые экземпляры приложений запускают OOM, а другие нет). Эта проблема OOM не относится к хранилищу по умолчанию (RocksDB), поскольку оно управляет своими данными на диске и использует память (RAM) только для целей кэширования. Но, опять же, вопрос о доступности приложения ортогонален безопасности данных (ваши данные в безопасности независимо от того, вылетает ли ваше приложение или нет).