look for help about task hooks of storm-compatibility

[Fangfengbin]

Hi 

@mjsax @StephanEwen @rmetzger
Can you give me some suggestion about my idea of how to transfer user-defined 
class code to task? 
Thank you very much!

Regards
Fengbin Fang

#################

Dear all,

I am work on task hooks of storm-compatibility. Storm support add hook through 
the Storm configuration using the "topology.auto.task.hooks" config. Users can 
use user-defined hooks class names as value of "topology.auto.task.hooks" of 
Configuration Map. These hooks are automatically registered in every spout or 
bolt.

Example like this:
public class Hooks extends BaseTaskHook {
    @Override
    public void boltExecute(BoltExecuteInfo info) {
        ...;
    }
}

public static void main(final String[] args) throws Exception {
    final FlinkTopologyBuilder builder = ExclamationTopology.buildTopology();
    Config config = new Config();
    config.put(Config.TOPOLOGY_AUTO_TASK_HOOKS, 
"org.apache.flink.stormcompatibility.util.Hook");

    // execute program locally
    final FlinkLocalCluster cluster = FlinkLocalCluster.getLocalCluster();
    cluster.submitTopology(topologyId, config, builder.createTopology()); }

My question is how to transfer user-defined class code to task?
Now I have a idea. I want to create user-defined hook object in FlinkClient and 
serialize it into bytes, then put bytes into storm configuration instead of 
user-defined hooks class names.
So all spout/bout can get user-defined hook object from storm configuration. In 
this way, we can make a cleaner separation between the storm-compatibility and 
core code.
But storm BaseTaskHook has not implements Serializable!  So user-defined hooks 
class must implements Serializable. This is the difference.

Can you give me some suggestion? Thank you very much!!

Regards
Fengbin Fang