如何迭代Kafka Streams表的键值

Question

我不熟悉Kafka streams，我尝试通过keyValueStore迭代Kafka streams表中的项：

这个想法是使用KeyValueStore创建一个Ktable（我也尝试过globalKTable）。然后一个分离的线程负责读取KeyValueStore的内容，以便遍历每个键的最后一个值。

      val streamProperties: Properties = {
      val p = new Properties()
      p.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-application")
      p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, config.getStringList("kafka.bootstrap.servers").toList.mkString(","))
      p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String.getClass.getName)
      p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArray.getClass.getName)
      p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
      p
    }

    val builder: StreamsBuilder = new StreamsBuilder()
    import org.apache.kafka.streams.kstream.Materialized
    import org.apache.kafka.streams.state.KeyValueStore


    val globalTable = builder.table("test",
      Materialized
        .as[String, Array[Byte], KeyValueStore[org.apache.kafka.common.utils.Bytes, Array[Byte]]]("TestStore")
        .withCachingDisabled()
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.ByteArray())
    )

    val streams: KafkaStreams = new KafkaStreams(builder.build(), streamProperties)
    streams.start()

    val ex = new ScheduledThreadPoolExecutor(1)
    val task = new Runnable {
      def run() = {
        println("\n\n\n tick \n\n\n")
        try {
          val keyValueStore = streams.store(globalTable.queryableStoreName(), QueryableStoreTypes.keyValueStore())
          keyValueStore.all().toIterator.map { k =>
            print(k.key)
          }
        } catch {
          case _ => println("error")
        }
      }
    }
    val f = ex.scheduleAtFixedRate(task, 1, 10, TimeUnit.SECONDS)
  }
}

在线程中，即使我在主题“test”上生成项目，keyValueStore也会保持为空。

有什么我错过或不明白的吗？

Answer 1

缺少的一件事是状态目录位置配置：

p.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp")

如果没有它，Kafka流不会抛出异常，但像全局KTables这样有状态的东西将无法默默地工作。