[FLINK-2391]Fix Storm-compatibility FlinkTopologyBuilder.createTopology bug

[ffbin]

1.Error Scene:
Error happend in program like this:
builder.setSpout("source0", new Generator(pt), pt.getInt("sourceParallelism"));
builder.setBolt("sa", new RepartPassThroughBolt(pt), pt.getInt("sinkParallelism")).fieldsGrouping("source0", new Fields("id"));
builder.setBolt("sink", new Sink(pt), pt.getInt("sinkParallelism")).fieldsGrouping("sa", new Fields("id"));
final FlinkLocalCluster cluster = FlinkLocalCluster.getLocalCluster();
cluster.submitTopology("throughput", conf, builder.createTopology());
if the last bolt use fieldsGrouping, createTopology will throw NullPointerException.
2.Reason:
where get streaming group attribute index, it get downstream operator outputFields,this is error。Because the last bolt has no
outputFields, so the outputSchema of declarer in null and throw NullPointerException.
3.Modify:
Save operator declarer in one hashmap. Get producer declarer from hashmap when get upstream operator outputFields.

[mjsax]

typo: prodDeclarer (not procDeclarer)

[mjsax]

LGTM. Let's see if Travis passes. Before merging, you need to update your commit message to be meaningful.

[mjsax]

Your commit message is still not ok. You update the GitHub description, but not the commit message. It is still "modify". You need to do a git commit --amend -m "New commit message". The commit message should relate to the JIRA, ie, "[FLINK-2391] Storm-compatibility:method FlinkTopologyBuilder.createTopology() throws java.lang.NullPointerException"

[rmetzger]

By the way, you don't need to open another pull request for updating it.
Just force-push into the branch this PR is based on (Feature2).

[ffbin]

Thank you very much!

[mjsax]

+1 for merging

[StephanEwen]

Will merge this...

[mjsax]

Thanks for reporting the bug and providing a fix.