Storm 0.9.4 to Storm 1.0.1升级指南

  1. 升级pom文件依赖至storm-core 1.0.1
  2. 升级pom文件依赖至storm-kafka 1.0.1
  3. 替换所有的backtype至org.apache
  4. 替换所有的storm.kafka至org.apache.storm.kafka
  5. 替换所有的storm.trident至org.apache.storm.trident
  6. 替换所有的org.apache.thrift7至org.apache.storm.thrift
  7. 使用storm-kafka时,用到SpoutConfig的forceFromStart替换为ignoreZkOffsets
  8. StormSubmitter.submitTopology方法新增一个抛出异常AuthorizationException,处理即可。
  9. Storm-kafka依赖0.9.0.1的kafka-clients,如需使用0.8.1的kafka集群,请在pom文件中排除kafka-clients依赖,并重新加入正确版本,如下:

<dependency>
    <groupId>org.apache.storm</groupId>
    <artifactId>storm-kafka</artifactId>
    <version>1.0.1</version>
        <exclusions>
            <exclusion>
                <artifactId>kafka-clients</artifactId>
                <groupId>org.apache.kafka</groupId>
            </exclusion>
        </exclusions>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>0.8.1.1</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
        </exclusion>
        <exclusion>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
        </exclusion>
    </exclusions>

</dependency>

如果所使用的kafka依赖版本里没有对kafka-clients的依赖,请再加入该依赖(storm-kafka依赖于kafka-clients)

其他已知改变:

1.Storm-kafka中所有的KafkaBolt(包括Trident和普通),不再使用配置”kafka.broker.properties”或xxBolt.KAFKA_BROKER_PROPERTIES ( 也即是 ”kafka.broker.properties” )来传递Kafka Producer的配置,而是使用withProducerProperties方法传递该配置(TridentKafkaState使用TridentKafkaStateFactory的该方法实现)。必要的三个配置为:bootstrap.servers,key.serializer,value.serializer。

2.org.apache.storm.tuple.Tuple类中的方法变更:已下方法被弃用,

@Deprecated

public GlobalStreamId getSourceGlobalStreamid();

新增方法:

public GlobalStreamId getSourceGlobalStreamId();

故代码升级时,需要Tuple的子类需要实现该方法。

3.由于storm 1.0.1的lib中删除了logback的包,故项目中用到的logback的,需要在pom文件中增加依赖。

    <dependency>

           <groupId>ch.qos.logback</groupId>

           <artifactId>logback-classic</artifactId>

           <version>1.1.6</version>

   </dependency>

4.org.apache.storm.spout.Scheme 中的函数List<Object> deserialize(ByteBuffer ser)的参数由byte[]转换为ByteBuffer。使用byte[] ser =org.apache.storm.utils.Utils.toByteArray(string)做转换

可能出现的问题:

  1. ERROR o.a.k.c.p.i.Sender – Uncaught error in kafka producer I/O thread: org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'throttle_time_ms': java.nio.BufferUnderflowException: kafka-clients版本不对,用0.9的clients往0.8的kafka上写数据就会有这个问题,排除方式见上。

注意事项:

  1. 由于storm 1.0中删除了logback的依赖,改用log4j2,故项目中用到的logback的,需要在pom文件中增加依赖。但增加了logback依赖后,slf4j会检测到多个底层log库,抛出Multiple Binding 警告,可能出现日志配置失效的情况。建议所有应用方使用slf4j,不要绕过slf4j直接使用底层日志库。http://www.slf4j.org/manual.html