I'm in the process of upgrading our Storm code from 0.10.0 to 1.0.0 and I've run into an issue with TransactionalTridentKafkaSpout. When running one of our topologies I'm getting the following exception:
Caused by: java.lang.ClassCastException: java.util.ArrayList cannot be cast to java.lang.Integer at org.apache.storm.trident.spout.PartitionedTridentSpoutExecutor$Coordinator.initializeTransaction(PartitionedTridentSpoutExecutor.java:55) ~[storm-core-1.0.0.jar:1.0.0] at org.apache.storm.trident.spout.PartitionedTridentSpoutExecutor$Coordinator.initializeTransaction(PartitionedTridentSpoutExecutor.java:43) ~[storm-core-1.0.0.jar:1.0.0] at org.apache.storm.trident.spout.TridentSpoutCoordinator.execute(TridentSpoutCoordinator.java:70) ~[storm-core-1.0.0.jar:1.0.0] at org.apache.storm.topology.BasicBoltExecutor.execute(BasicBoltExecutor.java:50) ~[storm-core-1.0.0.jar:1.0.0] The issue appears to be caused by a change in PartitionedTridentSpoutExecutor between the two versions, specifically this method: 1.0.0 - https://github.com/apache/storm/blob/v1.0.0/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java#L51 public Integer initializeTransaction(long txid, Integer prevMetadata, Integer currMetadata) { if(currMetadata!=null) { return currMetadata; } else { return _coordinator.getPartitionsForBatch(); } } 0.10.0 - https://github.com/apache/storm/blob/v0.10.0/storm-core/src/jvm/storm/trident/spout/PartitionedTridentSpoutExecutor.java#L51 public Object initializeTransaction(long txid, Object prevMetadata, Object currMetadata) { if(currMetadata!=null) { return currMetadata; } else { return _coordinator.getPartitionsForBatch(); } } The OpaquePartitionedTridentSpoutExecutor still uses Object for the metadata. Is this a bug in PartitionedTridentSpoutExecutor that is breaking transactional spouts? Any help will be appreciated. Thanks!
