使apache storm拓扑使用kafka的最新偏移量

Question

我有一个kafkaspout，两个bolt用于处理数据，两个bolt用于在mongodb中存储处理过的数据

我正在使用apache flux创建拓扑，在那里我将数据从Kafka读取到spout。一切都运行得很好，但每次运行拓扑时，它都从一开始就处理kafka中的所有消息。并且一旦它处理了所有的消息,它就不会等待更多的消息和崩溃。

我如何使Storm拓扑只处理最新的消息。

这是我的拓扑文件.yaml

name: "kafka-topology"

components:
# MongoDB mapper
  - id: "block-mapper"
    className: "org.apache.storm.mongodb.common.mapper.SimpleMongoMapper"
    configMethods:
      - name: "withFields"
        args: # The following are the tuple fields to map to a MongoDB document
          - ["block"]
# MongoDB mapper
  - id: "transaction-mapper"
    className: "org.apache.storm.mongodb.common.mapper.SimpleMongoMapper"
    configMethods:
      - name: "withFields"
        args: # The following are the tuple fields to map to a MongoDB document
          - ["transaction"]

  - id: "stringScheme"
    className: "org.apache.storm.kafka.StringScheme"

  - id: "stringMultiScheme"
    className: "org.apache.storm.spout.SchemeAsMultiScheme"
    constructorArgs:
      - ref: "stringScheme"

  - id: "zkHosts"
    className: "org.apache.storm.kafka.ZkHosts"
    constructorArgs:
      - "172.25.33.191:2181"

  - id: "spoutConfig"
    className: "org.apache.storm.kafka.SpoutConfig"
    constructorArgs:
      # brokerHosts
      - ref: "zkHosts"
      # topic
      - "blockdata"
      # zkRoot
      - ""
      # id
      - "myId"
    properties:
      - name: "scheme"
        ref: "stringMultiScheme"
      - name: "ignoreZkOffsets"
        value: flase



config:
  topology.workers: 1
  # ...

# spout definitions
spouts:
  - id: "kafka-spout"
    className: "org.apache.storm.kafka.KafkaSpout"
    constructorArgs:
      - ref: "spoutConfig"
    parallelism: 1

# bolt definitions
bolts:
  - id: "blockprocessing-bolt"
    className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
    constructorArgs:
      # command line
      - ["python", "process-bolt.py"]
      # output fields
      - ["block"]
    parallelism: 1
    # ...
  - id: "transprocessing-bolt"
    className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
    constructorArgs:
      # command line
      - ["python", "trans-bolt.py"]
      # output fields
      - ["transaction"]
    parallelism: 1
    # ...

  - id: "mongoBlock-bolt"
    className: "org.apache.storm.mongodb.bolt.MongoInsertBolt"
    constructorArgs:
      - "mongodb://172.25.33.205:27017/testdb"
      - "block"
      - ref: "block-mapper"
    parallelism: 1
    # ...

  - id: "mongoTrans-bolt"
    className: "org.apache.storm.mongodb.bolt.MongoInsertBolt"
    constructorArgs:
      - "mongodb://172.25.33.205:27017/testdb"
      - "transaction"
      - ref: "transaction-mapper"
    parallelism: 1
    # ...



  - id: "log"
    className: "org.apache.storm.flux.wrappers.bolts.LogInfoBolt"
    parallelism: 1
    # ...

#stream definitions
# stream definitions define connections between spouts and bolts.
# note that such connections can be cyclical
# custom stream groupings are also supported

streams:


  - name: "kafka --> block-Processing" # name isn't used (placeholder for logging, UI, etc.)
    from: "kafka-spout"
    to: "blockprocessing-bolt"
    grouping:
      type: SHUFFLE

  - name: "kafka --> transaction-processing" # name isn't used (placeholder for logging, UI, etc.)
    from: "kafka-spout"
    to: "transprocessing-bolt"
    grouping:
      type: SHUFFLE

  - name: "block --> mongo"
    from: "blockprocessing-bolt"
    to: "mongoBlock-bolt"
    grouping:
      type: SHUFFLE

  - name: "transaction --> mongo"
    from: "transprocessing-bolt"
    to: "mongoTrans-bolt"
    grouping:
      type: SHUFFLE

 - id: "spoutConfig"
className: "org.apache.storm.kafka.SpoutConfig"
constructorArgs:
  - ref: "zkHosts"
  - "blockdata"
  - ""
  - "myId"
properties:
  - name: "scheme"
    ref: "stringMultiScheme"
  - name: "startOffsetTime"
    ref: "EarliestTime"

  - name: "forceFromStart"
    value: false

Answer 1

您需要将startOffsetTime设置为kafka.api.offSetRequest.LatestTime。正如您在https://github.com/apache/storm/tree/64af629a19a82591dbf3428f7fd6b02f39e0723f/external/storm-kafka#kafkaconfig上看到的，默认设置将转到可用的最早偏移量。

你碰到的例外情况似乎不相关。这看起来像是馆长和动物园管理员的不相容。

编辑：我认为您正在访问这个问题https://issues.apache.org/jira/browse/storm-2978。1.2.2应该很快就会发布，请在发布后尝试升级。