{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/45733908","id":45733908,"diff_hunk":"@@ -15,75 +16,468 @@\n  * See the License for the specific language governing permissions and\n  * limitations under the License.\n  */\n-\n package org.apache.flink.storm.api;\n \n+import backtype.storm.generated.ComponentCommon;\n+import backtype.storm.generated.GlobalStreamId;\n+import backtype.storm.generated.Grouping;\n import backtype.storm.generated.StormTopology;\n+import backtype.storm.topology.IRichBolt;\n+import backtype.storm.topology.IRichSpout;\n+import backtype.storm.topology.IRichStateSpout;\n+import backtype.storm.topology.TopologyBuilder;\n+import backtype.storm.tuple.Fields;\n+import com.google.common.base.Preconditions;\n import org.apache.flink.api.common.JobExecutionResult;\n+import org.apache.flink.api.common.typeinfo.TypeInformation;\n+import org.apache.flink.api.java.tuple.Tuple;\n+import org.apache.flink.api.java.typeutils.TypeExtractor;\n+import org.apache.flink.storm.util.SplitStreamMapper;\n+import org.apache.flink.storm.util.SplitStreamType;\n+import org.apache.flink.storm.util.StormStreamSelector;\n+import org.apache.flink.storm.wrappers.BoltWrapper;\n+import org.apache.flink.storm.wrappers.BoltWrapperTwoInput;\n+import org.apache.flink.storm.wrappers.SpoutWrapper;\n+import org.apache.flink.streaming.api.datastream.DataStream;\n+import org.apache.flink.streaming.api.datastream.DataStreamSource;\n+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;\n+import org.apache.flink.streaming.api.datastream.SplitStream;\n import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;\n+import org.apache.flink.util.InstantiationUtil;\n+\n+import java.io.IOException;\n+import java.lang.reflect.Field;\n+import java.util.HashMap;\n+import java.util.HashSet;\n+import java.util.Iterator;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Map.Entry;\n+import java.util.Set;\n \n /**\n- * {@link FlinkTopology} mimics a {@link StormTopology} and is implemented in terms of a {@link\n- * StreamExecutionEnvironment} . In contrast to a regular {@link StreamExecutionEnvironment}, a {@link FlinkTopology}\n- * cannot be executed directly, but must be handed over to a {@link FlinkLocalCluster}, {@link FlinkSubmitter}, or\n- * {@link FlinkClient}.\n+ * {@link FlinkTopology} translates a {@link TopologyBuilder} to a Flink program.\n+ * <strong>CAUTION: {@link IRichStateSpout StateSpout}s are currently not supported.</strong>\n  */\n-public class FlinkTopology extends StreamExecutionEnvironment {\n+public class FlinkTopology {\n+\n+\t/** All declared streams and output schemas by operator ID */\n+\tprivate final HashMap<String, HashMap<String, Fields>> outputStreams = new HashMap<String, HashMap<String, Fields>>();\n+\t/** All spouts&bolts declarers by their ID */\n+\tprivate final HashMap<String, FlinkOutputFieldsDeclarer> declarers = new HashMap<String, FlinkOutputFieldsDeclarer>();\n+\n+\tprivate final HashMap<String, Set<Entry<GlobalStreamId, Grouping>>> unprocessdInputsPerBolt =\n+\t\t\tnew HashMap<String, Set<Entry<GlobalStreamId, Grouping>>>();\n+\n+\tfinal HashMap<String, HashMap<String, DataStream<Tuple>>> availableInputs = new HashMap<>();\n \n-\t/** The number of declared tasks for the whole program (ie, sum over all dops) */\n-\tprivate int numberOfTasks = 0;\n+\tprivate final TopologyBuilder builder;\n \n-\tpublic FlinkTopology() {\n-\t\t// Set default parallelism to 1, to mirror Storm default behavior\n-\t\tsuper.setParallelism(1);\n+\t// needs to be a class member for internal testing purpose\n+\tprivate final StormTopology stormTopology;\n+\n+\tprivate final Map<String, IRichSpout> spouts;\n+\tprivate final Map<String, IRichBolt> bolts;\n+\n+\tprivate final StreamExecutionEnvironment env;\n+\n+\tprivate FlinkTopology(TopologyBuilder builder) {\n+\t\tthis.builder = builder;\n+\t\tthis.stormTopology = builder.createTopology();\n+\t\t// extract the spouts and bolts\n+\t\tthis.spouts = getPrivateField(\"_spouts\");\n+\t\tthis.bolts = getPrivateField(\"_bolts\");\n+\n+\t\tthis.env = StreamExecutionEnvironment.getExecutionEnvironment();\n+\n+\t\t// Kick off the translation immediately\n+\t\ttranslateTopology();\n \t}\n \n \t/**\n-\t * Is not supported. In order to execute use {@link FlinkLocalCluster}, {@link FlinkSubmitter}, or {@link\n-\t * FlinkClient}.\n \t *\n-\t * @throws UnsupportedOperationException\n-\t * \t\tat every invocation\n+\t * Creates a Flink program that uses the specified spouts and bolts.\n+\t * @param stormBuilder The storm topology builder to use for creating the Flink topology.\n+\t * @return A Flink Topology which may be executed.\n \t */\n-\t@Override\n-\tpublic JobExecutionResult execute() throws Exception {\n-\t\tthrow new UnsupportedOperationException(\n-\t\t\t\t\"A FlinkTopology cannot be executed directly. Use FlinkLocalCluster, FlinkSubmitter, or FlinkClient \" +\n-\t\t\t\t\"instead.\");\n+\tpublic static FlinkTopology createTopology(TopologyBuilder stormBuilder) {\n+\t\treturn new FlinkTopology(stormBuilder);\n \t}\n \n \t/**\n-\t * Is not supported. In order to execute use {@link FlinkLocalCluster}, {@link FlinkSubmitter} or {@link\n-\t * FlinkClient}.\n-\t *\n-\t * @throws UnsupportedOperationException\n-\t * \t\tat every invocation\n+\t * Returns the underlying Flink ExecutionEnvironment for the Storm topology.\n+\t * @return The contextual environment.\n \t */\n-\t@Override\n-\tpublic JobExecutionResult execute(final String jobName) throws Exception {\n-\t\tthrow new UnsupportedOperationException(\n-\t\t\t\t\"A FlinkTopology cannot be executed directly. Use FlinkLocalCluster, FlinkSubmitter, or FlinkClient \" +\n-\t\t\t\t\"instead.\");\n+\tpublic StreamExecutionEnvironment getExecutionEnvironment() {\n+\t\treturn this.env;\n \t}\n \n \t/**\n-\t * Increased the number of declared tasks of this program by the given value.\n-\t *\n-\t * @param dop\n-\t * \t\tThe dop of a new operator that increases the number of overall tasks.\n+\t * Directly executes the Storm topology based on the current context (local when in IDE and\n+\t * remote when executed thorugh ./bin/flink).\n+\t * @return The execution result\n+\t * @throws Exception\n \t */\n-\tpublic void increaseNumberOfTasks(final int dop) {\n-\t\tassert (dop > 0);\n-\t\tthis.numberOfTasks += dop;\n+\tpublic JobExecutionResult execute() throws Exception {\n+\t\treturn env.execute();\n+\t}\n+\n+\n+\t@SuppressWarnings(\"unchecked\")\n+\tprivate <T> Map<String, T> getPrivateField(String field) {\n+\t\ttry {\n+\t\t\tField f = builder.getClass().getDeclaredField(field);\n+\t\t\tf.setAccessible(true);\n+\t\t\treturn copyObject((Map<String, T>) f.get(builder));\n+\t\t} catch (NoSuchFieldException | IllegalAccessException e) {\n+\t\t\tthrow new RuntimeException(\"Couldn't get \" + field + \" from TopologyBuilder\", e);\n+\t\t}\n+\t}\n+\n+\tprivate <T> T copyObject(T object) {\n+\t\ttry {\n+\t\t\treturn InstantiationUtil.deserializeObject(\n+\t\t\t\t\tInstantiationUtil.serializeObject(object),\n+\t\t\t\t\tgetClass().getClassLoader()\n+\t\t\t);\n+\t\t} catch (IOException | ClassNotFoundException e) {\n+\t\t\tthrow new RuntimeException(\"Failed to copy object.\");\n+\t\t}\n \t}\n \n \t/**\n-\t * Return the number or required tasks to execute this program.\n-\t *\n-\t * @return the number or required tasks to execute this program\n+\t * Creates a Flink program that uses the specified spouts and bolts.\n \t */\n-\tpublic int getNumberOfTasks() {\n-\t\treturn this.numberOfTasks;\n+\tprivate void translateTopology() {\n+\n+\t\tunprocessdInputsPerBolt.clear();\n+\t\toutputStreams.clear();\n+\t\tdeclarers.clear();\n+\t\tavailableInputs.clear();\n+\n+\t\t// Storm defaults to parallelism 1\n+\t\tenv.setParallelism(1);\n+\n+\t\t/* Translation of topology */\n+\n+\n+\t\tfor (final Entry<String, IRichSpout> spout : spouts.entrySet()) {\n+\t\t\tfinal String spoutId = spout.getKey();\n+\t\t\tfinal IRichSpout userSpout = spout.getValue();\n+\n+\t\t\tfinal FlinkOutputFieldsDeclarer declarer = new FlinkOutputFieldsDeclarer();\n+\t\t\tuserSpout.declareOutputFields(declarer);\n+\t\t\tfinal HashMap<String,Fields> sourceStreams = declarer.outputStreams;\n+\t\t\tthis.outputStreams.put(spoutId, sourceStreams);\n+\t\t\tdeclarers.put(spoutId, declarer);\n+\n+\n+\t\t\tfinal HashMap<String, DataStream<Tuple>> outputStreams = new HashMap<String, DataStream<Tuple>>();\n+\t\t\tfinal DataStreamSource<?> source;\n+\n+\t\t\tif (sourceStreams.size() == 1) {\n+\t\t\t\tfinal SpoutWrapper<Tuple> spoutWrapperSingleOutput = new SpoutWrapper<Tuple>(userSpout);\n+\t\t\t\tspoutWrapperSingleOutput.setStormTopology(stormTopology);\n+\n+\t\t\t\tfinal String outputStreamId = (String) sourceStreams.keySet().toArray()[0];\n+\n+\t\t\t\tDataStreamSource<Tuple> src = env.addSource(spoutWrapperSingleOutput, spoutId,\n+\t\t\t\t\t\tdeclarer.getOutputType(outputStreamId));\n+\n+\t\t\t\toutputStreams.put(outputStreamId, src);\n+\t\t\t\tsource = src;\n+\t\t\t} else {\n+\t\t\t\tfinal SpoutWrapper<SplitStreamType<Tuple>> spoutWrapperMultipleOutputs = new SpoutWrapper<SplitStreamType<Tuple>>(\n+\t\t\t\t\t\tuserSpout);\n+\t\t\t\tspoutWrapperMultipleOutputs.setStormTopology(stormTopology);\n+\n+\t\t\t\t@SuppressWarnings({ \"unchecked\", \"rawtypes\" })\n+\t\t\t\tDataStreamSource<SplitStreamType<Tuple>> multiSource = env.addSource(\n+\t\t\t\t\t\tspoutWrapperMultipleOutputs, spoutId,\n+\t\t\t\t\t\t(TypeInformation) TypeExtractor.getForClass(SplitStreamType.class));\n+\n+\t\t\t\tSplitStream<SplitStreamType<Tuple>> splitSource = multiSource\n+\t\t\t\t\t\t.split(new StormStreamSelector<Tuple>());\n+\t\t\t\tfor (String streamId : sourceStreams.keySet()) {\n+\t\t\t\t\toutputStreams.put(streamId, splitSource.select(streamId).map(new SplitStreamMapper<Tuple>()));\n+\t\t\t\t}\n+\t\t\t\tsource = multiSource;\n+\t\t\t}\n+\t\t\tavailableInputs.put(spoutId, outputStreams);\n+\n+\t\t\tfinal ComponentCommon common = stormTopology.get_spouts().get(spoutId).get_common();\n+\t\t\tif (common.is_set_parallelism_hint()) {\n+\t\t\t\tint dop = common.get_parallelism_hint();\n+\t\t\t\tsource.setParallelism(dop);\n+\t\t\t} else {\n+\t\t\t\tcommon.set_parallelism_hint(1);\n+\t\t\t}\n+\t\t}\n+\n+\t\t/**\n+\t\t* 1. Connect all spout streams with bolts streams\n+\t\t* 2. Then proceed with the bolts stream already connected\n+\t\t*\n+\t\t*  Because we do not know the order in which an iterator steps over a set, we might process a consumer before\n+\t\t* its producer\n+\t\t* ->thus, we might need to repeat multiple times\n+\t\t*/\n+\t\tboolean makeProgress = true;\n+\t\twhile (bolts.size() > 0) {\n+\t\t\tif (!makeProgress) {\n+\t\t\t\tthrow new RuntimeException(\n+\t\t\t\t\t\t\"Unable to build Topology. Could not connect the following bolts: \"\n+\t\t\t\t\t\t\t\t+ bolts.keySet());\n+\t\t\t}\n+\t\t\tmakeProgress = false;\n+\n+\t\t\tfinal Iterator<Entry<String, IRichBolt>> boltsIterator = bolts.entrySet().iterator();\n+\t\t\twhile (boltsIterator.hasNext()) {\n+\n+\t\t\t\tfinal Entry<String, IRichBolt> bolt = boltsIterator.next();\n+\t\t\t\tfinal String boltId = bolt.getKey();\n+\t\t\t\tfinal IRichBolt userBolt = copyObject(bolt.getValue());\n+\n+\t\t\t\tfinal ComponentCommon common = stormTopology.get_bolts().get(boltId).get_common();\n+\n+\t\t\t\tSet<Entry<GlobalStreamId, Grouping>> unprocessedBoltInputs = unprocessdInputsPerBolt.get(boltId);\n+\t\t\t\tif (unprocessedBoltInputs == null) {\n+\t\t\t\t\tunprocessedBoltInputs = new HashSet<>();\n+\t\t\t\t\tunprocessedBoltInputs.addAll(common.get_inputs().entrySet());\n+\t\t\t\t\tunprocessdInputsPerBolt.put(boltId, unprocessedBoltInputs);\n+\t\t\t\t}\n+\n+\t\t\t\t// check if all inputs are available\n+\t\t\t\tfinal int numberOfInputs = unprocessedBoltInputs.size();\n+\t\t\t\tint inputsAvailable = 0;\n+\t\t\t\tfor (Entry<GlobalStreamId, Grouping> entry : unprocessedBoltInputs) {\n+\t\t\t\t\tfinal String producerId = entry.getKey().get_componentId();\n+\t\t\t\t\tfinal String streamId = entry.getKey().get_streamId();\n+\t\t\t\t\tfinal HashMap<String, DataStream<Tuple>> streams = availableInputs.get(producerId);\n+\t\t\t\t\tif (streams != null && streams.get(streamId) != null) {\n+\t\t\t\t\t\tinputsAvailable++;\n+\t\t\t\t\t}\n+\t\t\t\t}\n+\n+\t\t\t\tif (inputsAvailable != numberOfInputs) {\n+\t\t\t\t\t// traverse other bolts first until inputs are available\n+\t\t\t\t\tcontinue;\n+\t\t\t\t} else {\n+\t\t\t\t\tmakeProgress = true;\n+\t\t\t\t\tboltsIterator.remove();\n+\t\t\t\t}\n+\n+\t\t\t\tfinal Map<GlobalStreamId, DataStream<Tuple>> inputStreams = new HashMap<>(numberOfInputs);\n+\n+\t\t\t\tfor (Entry<GlobalStreamId, Grouping> input : unprocessedBoltInputs) {\n+\t\t\t\t\tfinal GlobalStreamId streamId = input.getKey();\n+\t\t\t\t\tfinal Grouping grouping = input.getValue();\n+\n+\t\t\t\t\tfinal String producerId = streamId.get_componentId();\n+\n+\t\t\t\t\tfinal Map<String, DataStream<Tuple>> producer = availableInputs.get(producerId);\n+\n+\t\t\t\t\tinputStreams.put(streamId, processInput(boltId, userBolt, streamId, grouping, producer));\n+\t\t\t\t}\n+\n+\t\t\t\tfinal Iterator<Entry<GlobalStreamId, DataStream<Tuple>>> iterator = inputStreams.entrySet().iterator();\n+\n+\t\t\t\tfinal Entry<GlobalStreamId, DataStream<Tuple>> firstInput = iterator.next();\n+\t\t\t\tGlobalStreamId streamId = firstInput.getKey();\n+\t\t\t\tDataStream<Tuple> inputStream = firstInput.getValue();\n+\n+\t\t\t\tfinal SingleOutputStreamOperator<?, ?> outputStream;\n+\n+\t\t\t\tswitch (numberOfInputs) {\n+\t\t\t\t\tcase 1:\n+\t\t\t\t\t\toutputStream = createOutput(boltId, userBolt, streamId, inputStream);\n+\t\t\t\t\t\tbreak;\n+\t\t\t\t\tcase 2:\n+\t\t\t\t\t\tEntry<GlobalStreamId, DataStream<Tuple>> secondInput = iterator.next();\n+\t\t\t\t\t\tGlobalStreamId streamId2 = secondInput.getKey();\n+\t\t\t\t\t\tDataStream<Tuple> inputStream2 = secondInput.getValue();\n+\t\t\t\t\t\toutputStream = createOutput(boltId, userBolt, streamId, inputStream, streamId2, inputStream2);\n+\t\t\t\t\t\tbreak;\n+\t\t\t\t\tdefault:\n+\t\t\t\t\t\tthrow new UnsupportedOperationException(\"Don't know how to translate a bolt \"\n+\t\t\t\t\t\t\t\t+ boltId + \" with \" + numberOfInputs + \" inputs.\");\n+\t\t\t\t}\n+\n+\t\t\t\tif (common.is_set_parallelism_hint()) {\n+\t\t\t\t\tint dop = common.get_parallelism_hint();\n+\t\t\t\t\toutputStream.setParallelism(dop);\n+\t\t\t\t} else {\n+\t\t\t\t\tcommon.set_parallelism_hint(1);\n+\t\t\t\t}\n+\n+\t\t\t}\n+\t\t}\n \t}\n \n+\tprivate DataStream<Tuple> processInput(String boltId, IRichBolt userBolt,\n+\t\t\t\t\t\t\t\t\t\tGlobalStreamId streamId, Grouping grouping,\n+\t\t\t\t\t\t\t\t\t\tMap<String, DataStream<Tuple>> producer) {\n+\n+\t\tPreconditions.checkNotNull(userBolt);","path":"flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkTopology.java","position":354,"original_position":354,"commit_id":"59cc00cbde90dedff218b16ebfc3671a03b9c32d","original_commit_id":"613936cc7dce75e8dd811626cd2152b1f1383fe0","user":{"login":"mxm","id":837221,"avatar_url":"https://avatars.githubusercontent.com/u/837221?v=3","gravatar_id":"","url":"https://api.github.com/users/mxm","html_url":"https://github.com/mxm","followers_url":"https://api.github.com/users/mxm/followers","following_url":"https://api.github.com/users/mxm/following{/other_user}","gists_url":"https://api.github.com/users/mxm/gists{/gist_id}","starred_url":"https://api.github.com/users/mxm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mxm/subscriptions","organizations_url":"https://api.github.com/users/mxm/orgs","repos_url":"https://api.github.com/users/mxm/repos","events_url":"https://api.github.com/users/mxm/events{/privacy}","received_events_url":"https://api.github.com/users/mxm/received_events","type":"User","site_admin":false},"body":"Yes, but these checks can be executed during run time of the user program. Assert only checks these when debugging is enabled.","created_at":"2015-11-24T13:26:45Z","updated_at":"2015-11-24T13:26:45Z","html_url":"https://github.com/apache/flink/pull/1398#discussion_r45733908","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/1398","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/45733908"},"html":{"href":"https://github.com/apache/flink/pull/1398#discussion_r45733908"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/1398"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/1398","id":51638971,"html_url":"https://github.com/apache/flink/pull/1398","diff_url":"https://github.com/apache/flink/pull/1398.diff","patch_url":"https://github.com/apache/flink/pull/1398.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/1398","number":1398,"state":"open","locked":false,"title":"[FLINK-2837][storm] various improvements for Storm compatibility","user":{"login":"mxm","id":837221,"avatar_url":"https://avatars.githubusercontent.com/u/837221?v=3","gravatar_id":"","url":"https://api.github.com/users/mxm","html_url":"https://github.com/mxm","followers_url":"https://api.github.com/users/mxm/followers","following_url":"https://api.github.com/users/mxm/following{/other_user}","gists_url":"https://api.github.com/users/mxm/gists{/gist_id}","starred_url":"https://api.github.com/users/mxm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mxm/subscriptions","organizations_url":"https://api.github.com/users/mxm/orgs","repos_url":"https://api.github.com/users/mxm/repos","events_url":"https://api.github.com/users/mxm/events{/privacy}","received_events_url":"https://api.github.com/users/mxm/received_events","type":"User","site_admin":false},"body":"This pull request contains various fixes. Most prominently, the parsing logic for Storm topologies has been changed to support multiple inputs. The API has been redone slightly. Two new examples have been added.\r\n\r\n- refactor to use Storm's topology builder\r\n\r\n- remove FlinkTopologyBuilder\r\n\r\n- instantiate context-based StreamExecutionEnvironment (local or remote)\r\n\r\n- remove Flink and Storm behavior replicating classes\r\n\r\n- modify FlinkTopology to parse Storm topology directly\r\n\r\n- replace StormTestBase with StreamingTestBase\r\n\r\n- let the FiniteFileSpout finish in corner cases\r\n\r\n- expose taskId, fix off by one task id\r\n\r\n- add print example\r\n\r\n- FlinkTopologyBuilder changes (check if all inputs are available before processing)\r\n\r\n- correct package typo\r\n\r\n- support getter methods on TupleWrapper\r\n\r\n- two input support\r\n\r\n- add join example\r\n\r\n- update docs\r\n\r\n- use Flink file system access","created_at":"2015-11-24T10:09:02Z","updated_at":"2015-11-24T13:26:45Z","closed_at":null,"merged_at":null,"merge_commit_sha":"8f45df735b02f0da9e773de789ccc307c3a66d1c","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/1398/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/1398/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/1398/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/59cc00cbde90dedff218b16ebfc3671a03b9c32d","head":{"label":"mxm:storm-dev-dev","ref":"storm-dev-dev","sha":"59cc00cbde90dedff218b16ebfc3671a03b9c32d","user":{"login":"mxm","id":837221,"avatar_url":"https://avatars.githubusercontent.com/u/837221?v=3","gravatar_id":"","url":"https://api.github.com/users/mxm","html_url":"https://github.com/mxm","followers_url":"https://api.github.com/users/mxm/followers","following_url":"https://api.github.com/users/mxm/following{/other_user}","gists_url":"https://api.github.com/users/mxm/gists{/gist_id}","starred_url":"https://api.github.com/users/mxm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mxm/subscriptions","organizations_url":"https://api.github.com/users/mxm/orgs","repos_url":"https://api.github.com/users/mxm/repos","events_url":"https://api.github.com/users/mxm/events{/privacy}","received_events_url":"https://api.github.com/users/mxm/received_events","type":"User","site_admin":false},"repo":{"id":25965625,"name":"flink","full_name":"mxm/flink","owner":{"login":"mxm","id":837221,"avatar_url":"https://avatars.githubusercontent.com/u/837221?v=3","gravatar_id":"","url":"https://api.github.com/users/mxm","html_url":"https://github.com/mxm","followers_url":"https://api.github.com/users/mxm/followers","following_url":"https://api.github.com/users/mxm/following{/other_user}","gists_url":"https://api.github.com/users/mxm/gists{/gist_id}","starred_url":"https://api.github.com/users/mxm/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mxm/subscriptions","organizations_url":"https://api.github.com/users/mxm/orgs","repos_url":"https://api.github.com/users/mxm/repos","events_url":"https://api.github.com/users/mxm/events{/privacy}","received_events_url":"https://api.github.com/users/mxm/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/mxm/flink","description":"Mirror of Apache Flink","fork":true,"url":"https://api.github.com/repos/mxm/flink","forks_url":"https://api.github.com/repos/mxm/flink/forks","keys_url":"https://api.github.com/repos/mxm/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/mxm/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/mxm/flink/teams","hooks_url":"https://api.github.com/repos/mxm/flink/hooks","issue_events_url":"https://api.github.com/repos/mxm/flink/issues/events{/number}","events_url":"https://api.github.com/repos/mxm/flink/events","assignees_url":"https://api.github.com/repos/mxm/flink/assignees{/user}","branches_url":"https://api.github.com/repos/mxm/flink/branches{/branch}","tags_url":"https://api.github.com/repos/mxm/flink/tags","blobs_url":"https://api.github.com/repos/mxm/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/mxm/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/mxm/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/mxm/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/mxm/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/mxm/flink/languages","stargazers_url":"https://api.github.com/repos/mxm/flink/stargazers","contributors_url":"https://api.github.com/repos/mxm/flink/contributors","subscribers_url":"https://api.github.com/repos/mxm/flink/subscribers","subscription_url":"https://api.github.com/repos/mxm/flink/subscription","commits_url":"https://api.github.com/repos/mxm/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/mxm/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/mxm/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/mxm/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/mxm/flink/contents/{+path}","compare_url":"https://api.github.com/repos/mxm/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/mxm/flink/merges","archive_url":"https://api.github.com/repos/mxm/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/mxm/flink/downloads","issues_url":"https://api.github.com/repos/mxm/flink/issues{/number}","pulls_url":"https://api.github.com/repos/mxm/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/mxm/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/mxm/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/mxm/flink/labels{/name}","releases_url":"https://api.github.com/repos/mxm/flink/releases{/id}","created_at":"2014-10-30T10:24:48Z","updated_at":"2015-04-29T12:00:18Z","pushed_at":"2015-11-24T12:21:43Z","git_url":"git://github.com/mxm/flink.git","ssh_url":"git@github.com:mxm/flink.git","clone_url":"https://github.com/mxm/flink.git","svn_url":"https://github.com/mxm/flink","homepage":null,"size":68999,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"89b1d233254de024942c06700cedd86fff55970e","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":20587599,"name":"flink","full_name":"apache/flink","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/flink","description":"Mirror of Apache Flink","fork":false,"url":"https://api.github.com/repos/apache/flink","forks_url":"https://api.github.com/repos/apache/flink/forks","keys_url":"https://api.github.com/repos/apache/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/flink/teams","hooks_url":"https://api.github.com/repos/apache/flink/hooks","issue_events_url":"https://api.github.com/repos/apache/flink/issues/events{/number}","events_url":"https://api.github.com/repos/apache/flink/events","assignees_url":"https://api.github.com/repos/apache/flink/assignees{/user}","branches_url":"https://api.github.com/repos/apache/flink/branches{/branch}","tags_url":"https://api.github.com/repos/apache/flink/tags","blobs_url":"https://api.github.com/repos/apache/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/flink/languages","stargazers_url":"https://api.github.com/repos/apache/flink/stargazers","contributors_url":"https://api.github.com/repos/apache/flink/contributors","subscribers_url":"https://api.github.com/repos/apache/flink/subscribers","subscription_url":"https://api.github.com/repos/apache/flink/subscription","commits_url":"https://api.github.com/repos/apache/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/flink/contents/{+path}","compare_url":"https://api.github.com/repos/apache/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/flink/merges","archive_url":"https://api.github.com/repos/apache/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/flink/downloads","issues_url":"https://api.github.com/repos/apache/flink/issues{/number}","pulls_url":"https://api.github.com/repos/apache/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/flink/labels{/name}","releases_url":"https://api.github.com/repos/apache/flink/releases{/id}","created_at":"2014-06-07T07:00:10Z","updated_at":"2015-11-24T06:14:33Z","pushed_at":"2015-11-24T12:49:38Z","git_url":"git://github.com/apache/flink.git","ssh_url":"git@github.com:apache/flink.git","clone_url":"https://github.com/apache/flink.git","svn_url":"https://github.com/apache/flink","homepage":null,"size":85160,"stargazers_count":760,"watchers_count":760,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":508,"mirror_url":"git://git.apache.org/incubator-flink.git","open_issues_count":76,"forks":508,"open_issues":76,"watchers":760,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/1398"},"html":{"href":"https://github.com/apache/flink/pull/1398"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/1398"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/1398/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/1398/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/1398/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/59cc00cbde90dedff218b16ebfc3671a03b9c32d"}}}}