[jira] [Created] (STORM-1854) Trident transactional spouts are broken in 1.0.x

["Morrigan Jones (JIRA)" <ji...@apache.org>]

Morrigan Jones created STORM-1854:
-------------------------------------

             Summary: Trident transactional spouts are broken in 1.0.x
                 Key: STORM-1854
                 URL: https://issues.apache.org/jira/browse/STORM-1854
             Project: Apache Storm
          Issue Type: Bug
          Components: trident
    Affects Versions: 1.0.0, 1.0.1
            Reporter: Morrigan Jones


In the process of upgrading our Storm code from 0.10.0 to 1.0.0, 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();            
    }
}

This was introduced by: https://github.com/apache/storm/commit/9e4c3df17ffbc737210e606d3d8a9cdae8f86634

TransactionalTridentKafkaSpout uses List<GlobalPartitionInformation> for its metadata. Generally, transactional spouts should have metadata that is more complex than just an Integer. OpaquePartitionedTridentSpoutExecutor uses Object for its metadata and correctly handles the metadata used by OpaqueTridentKafkaSpout (List<GlobalPartitionInformation>).

It looks like reverting the metadata type for transactional spouts in PartitionedTridentSpoutExecutor should work, but I haven't tried this yet.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)