From 9db92f6714842af14a244544b6e34b2814e2565b Mon Sep 17 00:00:00 2001
From: Dan Herrera <whoahbot@gmail.com>
Date: Tue, 17 May 2016 14:12:57 -0700
Subject: [PATCH] [Storm-1728] TransactionalTridentKafkaSpout error

---
 .../spout/PartitionedTridentSpoutExecutor.java   | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java b/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java
index 8aa97283b6b..389a9298bc9 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/spout/PartitionedTridentSpoutExecutor.java
@@ -29,7 +29,7 @@
 import org.apache.storm.trident.topology.state.TransactionalState;
 
 
-public class PartitionedTridentSpoutExecutor implements ITridentSpout<Integer> {
+public class PartitionedTridentSpoutExecutor implements ITridentSpout<Object> {
     IPartitionedTridentSpout<Integer, ISpoutPartition, Object> _spout;
     
     public PartitionedTridentSpoutExecutor(IPartitionedTridentSpout<Integer, ISpoutPartition, Object> spout) {
@@ -40,7 +40,7 @@ public IPartitionedTridentSpout<Integer, ISpoutPartition, Object> getPartitioned
         return _spout;
     }
     
-    class Coordinator implements ITridentSpout.BatchCoordinator<Integer> {
+    class Coordinator implements ITridentSpout.BatchCoordinator<Object> {
         private IPartitionedTridentSpout.Coordinator<Integer> _coordinator;
         
         public Coordinator(Map conf, TopologyContext context) {
@@ -48,7 +48,7 @@ public Coordinator(Map conf, TopologyContext context) {
         }
         
         @Override
-        public Integer initializeTransaction(long txid, Integer prevMetadata, Integer currMetadata) {
+        public Object initializeTransaction(long txid, Object prevMetadata, Object currMetadata) {
             if(currMetadata!=null) {
                 return currMetadata;
             } else {
@@ -82,8 +82,8 @@ public EmitterPartitionState(RotatingTransactionalState s, ISpoutPartition p) {
         }
     }
     
-    class Emitter implements ITridentSpout.Emitter<Integer> {
-        private IPartitionedTridentSpout.Emitter<Integer, ISpoutPartition, Object> _emitter;
+    class Emitter implements ITridentSpout.Emitter<Object> {
+        private IPartitionedTridentSpout.Emitter _emitter;
         private TransactionalState _state;
         private Map<String, EmitterPartitionState> _partitionStates = new HashMap<>();
         private int _index;
@@ -100,7 +100,7 @@ public Emitter(String txStateId, Map conf, TopologyContext context) {
 
         
         @Override
-        public void emitBatch(final TransactionAttempt tx, final Integer coordinatorMeta,
+        public void emitBatch(final TransactionAttempt tx, final Object coordinatorMeta,
                 final TridentCollector collector) {
             if(_savedCoordinatorMeta == null || !_savedCoordinatorMeta.equals(coordinatorMeta)) {
                 List<ISpoutPartition> partitions = _emitter.getOrderedPartitions(coordinatorMeta);
@@ -150,12 +150,12 @@ public void close() {
     }    
 
     @Override
-    public ITridentSpout.BatchCoordinator<Integer> getCoordinator(String txStateId, Map conf, TopologyContext context) {
+    public ITridentSpout.BatchCoordinator getCoordinator(String txStateId, Map conf, TopologyContext context) {
         return new Coordinator(conf, context);
     }
 
     @Override
-    public ITridentSpout.Emitter<Integer> getEmitter(String txStateId, Map conf, TopologyContext context) {
+    public ITridentSpout.Emitter<Object> getEmitter(String txStateId, Map conf, TopologyContext context) {
         return new Emitter(txStateId, conf, context);
     }