You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by mjsax <gi...@git.apache.org> on 2015/04/06 18:11:29 UTC
[GitHub] flink pull request: Flink Storm compatibility
GitHub user mjsax opened a pull request:
https://github.com/apache/flink/pull/573
Flink Storm compatibility
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mjsax/flink flink-storm-compatibility
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/573.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #573
----
commit 1470b0f4af86d457d841afc037bd9583cdb9d1c1
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-02-24T18:39:27Z
fixed package statement
commit 425acb7a2717bde543ff28f794bd45a00ebdc2fb
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-02-25T10:25:54Z
Merge branch 'bug_fix_package_statement'
commit 9ee84ca31c05fb4192a288b5d236c519d162d3e2
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-03-10T07:46:32Z
Merge branch 'master' of https://github.com/apache/flink
commit adcac585233ff6249d8cfe10a0773f1f1b624c04
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-03-20T16:21:34Z
Merge branch 'master' of https://github.com/apache/flink
commit 8091facd05bb00a4d69499ba86c06ed2e928d9aa
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-03-22T00:56:38Z
Merge branch 'master' of https://github.com/apache/flink
commit a21f94747649174f63645c95f1b53731eb17c2ba
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-04-06T13:23:06Z
Merge branch 'master' of https://github.com/apache/flink
commit 104d21f13abcd18dbe6e070f6400e46d557b9cb8
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-02-26T08:50:30Z
initial commit for storm-compatibility module
commit de3a7a81aae7fa63fb7c95961b15b8906c3e2597
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-02-27T14:40:40Z
alpha version of a wrapper machanism to use Storm bolts within a Flink Streaming program
commit 86ad016461956ada3f4a84d229a42b79f8993adb
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-03-04T16:17:52Z
Added test for StormBoltWrapper and its utility classes
commit bd154124ae359e0b19c7e8629caaba5258f56cd6
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-03-16T18:03:42Z
added raw output type support
commit 2711b9fdeef739771e233d9a90fb338c9a9a7781
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-03-16T21:41:46Z
added StormSpoutWrapper including JUnit and ITCase
fixed bugs and improved existing tests
commit bbaa9b2d8f136b193a952cbae9488b43499ed90f
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-03-25T00:59:17Z
extracted Storm Spouts and Bolt to seperate package
improved StormFinitSpoutWrapper
added support for Storm sink-bolts
improved integration of Flink and Storm configuration
commit d8a9e1e57a85d772d891b8e765477c84c28bfd27
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-03-27T17:34:58Z
added API compatiblity classes
commit 70d8b49c8c5c7b235d50008447d14c821c00615f
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-04-01T07:51:25Z
added isRunning flag to spout wrappers
commit a0a4c4a8ab0e5abb580e5b1f4b7e8e131796a569
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-04-01T18:02:30Z
worked on LocalCluster shutdown for integration tests
commit ce932a51f9da275896afb0d6a73abc661d9fb1c9
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-04-02T13:37:58Z
Added Storm clients that connect to Flink
added a submit-remote-example
commit 1519d36e439d5d76d1e30cf54509a214f065eb8c
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-04-02T14:39:25Z
added new JUnit test
improved existing tests
renamed package org.apache.flink.stormcompatibility.operators to org.apache.flink.stormcompatibility.wrappers
commit 605de53a9990dbe2941261d1365654fe62a7c91d
Author: mjsax <mj...@informatik.hu-berlin.de>
Date: 2015-04-02T17:15:32Z
added missing license header
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30113258
--- Diff: flink-staging/flink-streaming/flink-storm-examples/src/main/java/org/apache/flink/stormcompatibility/wordcount/StormWordCountLocal.java ---
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.wordcount;
+
+import org.apache.flink.examples.java.wordcount.util.WordCountData;
+import org.apache.flink.stormcompatibility.api.FlinkLocalCluster;
+import org.apache.flink.stormcompatibility.api.FlinkTopologyBuilder;
+
+import backtype.storm.LocalCluster;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * Implements the "WordCount" program that computes a simple word occurrence histogram over text files in a streaming
+ * fashion. The program is constructed as a regular {@link StormTopology} and submitted to Flink for execution in the
+ * same way as to a Storm {@link LocalCluster}.
+ *
+ * This example shows how to run program directly within Java, thus it cannot be used to submit a {@link StormTopology}
+ * via Flink command line clients (ie, bin/flink).
+ *
+ * <p>
+ * The input is a plain text file with lines separated by newline characters.
+ *
+ * <p>
+ * Usage: <code>WordCount <text path> <result path></code><br>
+ * If no parameters are provided, the program is run with default data from {@link WordCountData}.
+ *
+ * <p>
+ * This example shows how to:
+ * <ul>
+ * <li>run a regular Storm program locally on Flink
+ * </ul>
+ */
+public class StormWordCountLocal {
+ public final static String topologyId = "Streaming WordCount";
+
+ // *************************************************************************
+ // PROGRAM
+ // *************************************************************************
+
+ public static void main(final String[] args) throws Exception {
+
+ if(!WordCountTopology.parseParameters(args)) {
+ return;
+ }
+
+ // build Topology the Storm way
+ final FlinkTopologyBuilder builder = WordCountTopology.buildTopology();
+
+ // execute program locally
+ final FlinkLocalCluster cluster = FlinkLocalCluster.getLocalCluster();
+ cluster.submitTopology(topologyId, null, builder.createTopology());
+
+ Utils.sleep(5 * 1000);
--- End diff --
Why is it necessary to sleep in the examples? Isn't there a way to submit the topology in a blocking way?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/573
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-90245431
Thanks, let me do another pass over the code later this week.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808183
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.ClusterSummary;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link backtype.storm.generated.Nimbus.Client
+ * Nimbus.Client} at once, to interact with Flink's JobManager instead of Storm's Nimbus.
+ *
+ * @author mjsax
+ */
+public class FlinkClient {
+ /**
+ * Configuration key to specify the location of the file storm-core-0.9.4.jar (for org.apache.storm).
+ */
+ public final static String FLINK_STROM_CORE_JAR_LOCATION = "flink.storm.jar-location.storm-core";
+ /**
+ * Configuration key to specify the location of the file json-simple-1.1.jar (for com.googlecode.json-simple).
+ */
+ public final static String FLINK_STROM_JSON_SIMPLE_JAR_LOCATION = "flink.storm.jar-location.json-simple";
+
+
+
+ /**
+ * The users specified configuration.
+ */
+ private final Map<?, ?> config;
+ /**
+ * The jobmanager's host name.
+ */
+ private final String jobManagerHost;
+ /**
+ * The jobmanager's rpc port.
+ */
+ private final int jobManagerPort;
+ /**
+ * The user specified timeout in milliseconds.
+ */
+ private final String timeout;
+ /**
+ * Maps the names of running topologies to the corresponding Flink {@link JobID}.
+ */
+ private final Map<String, JobID> topologyNameToJobId = new HashMap<String, JobID>();
+ /**
+ * Maps the names of running topologies to the corresponding {@link FlinkTopology}.
+ */
+ private final Map<String, FlinkTopology> topologies = new HashMap<String, FlinkTopology>();
+
+
+
+ /*
+ * The following methods are derived from "backtype.storm.utils.NimbusClient"
+ */
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port) {
+ this(conf, host, port, null);
+ }
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ * @param timeout
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port, final Integer timeout) {
+ this.config = conf;
+ this.jobManagerHost = host;
+ this.jobManagerPort = port;
+ if(timeout != null) {
+ this.timeout = timeout + " ms";
+ } else {
+ this.timeout = null;
+ }
+ }
+
+
+
+ /**
+ * Returns a {@link FlinkClient} that uses the configured {@link Config#NIMBUS_HOST} and
+ * {@link Config#NIMBUS_THRIFT_PORT} as JobManager address.
+ *
+ * @param conf
+ * Configuration that contains the jobmanager's hostname and port.
+ *
+ * @return A configured {@link FlinkClient}.
+ */
+ public static FlinkClient getConfiguredClient(@SuppressWarnings("rawtypes") final Map conf) {
+ final String nimbusHost = (String)conf.get(Config.NIMBUS_HOST);
+ final int nimbusPort = Utils.getInt(conf.get(Config.NIMBUS_THRIFT_PORT)).intValue();
+ return new FlinkClient(conf, nimbusHost, nimbusPort);
+ }
+
+
+
+ /**
+ * Return a reference to itself.
+ *
+ * {@link FlinkClient} mimics both, {@link NimbusClient} and {@link backtype.storm.generated.Nimubs.Client
+ * Nimbus.Client}, at once.
+ *
+ * @return A reference to itself.
+ */
+ public FlinkClient getClient() {
+ return this;
+ }
+
+ public void close() { /* nothing to do */}
+
+
+ /*
+ * The following methods are derived from "backtype.storm.generated.Nimubs.Client"
+ */
+
+ /**
+ * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
+ * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
+ */
+ public void submitTopology(final String name, final String uploadedJarLocation, final String jsonConf, final FlinkTopology topology)
+ throws AlreadyAliveException, InvalidTopologyException {
+ this.submitTopologyWithOpts(name, uploadedJarLocation, jsonConf, topology, null);
+ }
+
+ /**
+ * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
+ * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
+ */
+ @SuppressWarnings("unused")
+ public void submitTopologyWithOpts(final String name, final String uploadedJarLocation, final String jsonConf, final FlinkTopology topology, final SubmitOptions options)
+ throws AlreadyAliveException, InvalidTopologyException {
+
+ // TODO improve by checking job manager, if running
+ if(this.topologyNameToJobId.containsKey(name)) {
+ throw new AlreadyAliveException();
+ }
+
+ // TODO refactor and combine shared code (copied from RemoveStreamEnvironment)
+ try {
+ final JobGraph jobGraph = topology.getStreamGraph().getJobGraph(name);
+
+ // TODO improve error handling for missing config values and uploadedJarLocation==null
+ final String stormCoreJar = (String)this.config.get(FLINK_STROM_CORE_JAR_LOCATION);
+ final String jsonSimpleJar = (String)this.config.get(FLINK_STROM_JSON_SIMPLE_JAR_LOCATION);
+ final String[] jarFileNames = new String[] {stormCoreJar, jsonSimpleJar, uploadedJarLocation};
+
+ final List<File> jarFiles = new ArrayList<File>();
+ for(int i = 0; i < jarFileNames.length; i++) {
+ final File file = new File(jarFileNames[i]);
+ try {
+ JobWithJars.checkJarFile(file);
+ } catch(final IOException e) {
+ throw new RuntimeException("Problem with jar file " + file.getAbsolutePath(), e);
+ }
+ jarFiles.add(file);
+ }
+
+ for(final File file : jarFiles) {
+ jobGraph.addJar(new Path(file.getAbsolutePath()));
+ }
+
+ final Configuration configuration = jobGraph.getJobConfiguration();
+ final Client client = new Client(new InetSocketAddress(this.jobManagerHost, this.jobManagerPort),
+ configuration, JobWithJars.buildUserCodeClassLoader(jarFiles, JobWithJars.class.getClassLoader()), -1);
+
+ try {
+ final JobSubmissionResult result = client.run(jobGraph, false);
+ this.topologyNameToJobId.put(name, result.getJobID());
+ this.topologies.put(name, topology);
+ } catch(final ProgramInvocationException e) {
+ throw new RuntimeException("Cannot execute job due to ProgramInvocationException", e);
+ }
+
+ } catch(final Exception e) {
+ e.printStackTrace();
+ }
--- End diff --
This happens on the client side, but I would still log it instead of printing stack trace.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27811153
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java ---
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.wrappers;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.flink.api.java.tuple.Tuple25;
+import org.apache.flink.stormcompatibility.api.FlinkTopologyContext;
+import org.apache.flink.streaming.api.streamvertex.StreamingRuntimeContext;
+
+import backtype.storm.generated.Bolt;
+import backtype.storm.generated.ComponentCommon;
+import backtype.storm.generated.SpoutSpec;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.IComponent;
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.IRichSpout;
+
+
+
+
+
+/**
+ * {@link StormWrapperSetupHelper} is an helper class used by {@link AbstractStormSpoutWrapper} or
+ * {@link StormBoltWrapper}.
+ *
+ * @author mjsax
+ */
+class StormWrapperSetupHelper {
+
+ /**
+ * Computes the number of output attributes used by a {@link AbstractStormSpoutWrapper} or {@link StormBoltWrapper}.
+ * Returns zero for raw output type or a value within range [1;25] for output type {@link Tuple1} to {@link Tuple25}
+ * . In case of a data sink, {@code -1} is returned. .
+ *
+ * @param spoutOrBolt
+ * The Storm {@link IRichSpout spout} or {@link IRichBolt bolt} to be used.
+ * @param rawOutput
+ * Set to {@code true} if a single attribute output stream, should not be of type {@link Tuple1} but be
+ * of a raw type.
+ * @return The number of attributes to be used.
+ * @throws IllegalArgumentException
+ * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if
+ * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range
+ * [1;25].
+ */
+ public static int getNumberOfAttributes(final IComponent spoutOrBolt, final boolean rawOutput)
+ throws IllegalArgumentException {
+ final StormOutputFieldsDeclarer declarer = new StormOutputFieldsDeclarer();
+ spoutOrBolt.declareOutputFields(declarer);
+
+ final int declaredNumberOfAttributes = declarer.getNumberOfAttributes();
+
+ if(declaredNumberOfAttributes == -1) {
+ return -1;
+ }
+
+ if((declaredNumberOfAttributes < 1) || (declaredNumberOfAttributes > 25)) {
+ throw new IllegalArgumentException(
+ "Provided bolt declares non supported number of output attributes. Must be in range [1;25] but was "
+ + declaredNumberOfAttributes);
+ }
+
+ if(rawOutput) {
+ if(declaredNumberOfAttributes > 1) {
+ throw new IllegalArgumentException(
+ "Ouput type is requested to be raw type, but provided bolt declares more then one output attribute.");
+
+ }
+ return 0;
+ }
+
+ return declaredNumberOfAttributes;
+ }
+
+ /**
+ * TODO
+ */
+ public static TopologyContext convertToTopologyContext(final StreamingRuntimeContext context, final boolean spoutOrBolt) {
+ final Integer taskId = new Integer(1 + context.getIndexOfThisSubtask());
+
+ final Map<Integer, String> taskToComponents = new HashMap<Integer, String>();
+ taskToComponents.put(taskId, context.getTaskName());
+
+ final ComponentCommon common = new ComponentCommon();
+ common.set_parallelism_hint(context.getNumberOfParallelSubtasks());
+
+ final Map<String, Bolt> bolts = new HashMap<String, Bolt>();
+ final Map<String, SpoutSpec> spoutSpecs = new HashMap<String, SpoutSpec>();
+
+ if(spoutOrBolt) {
+ spoutSpecs.put(context.getTaskName(), new SpoutSpec(null, common));
+ } else {
+ bolts.put(context.getTaskName(), new Bolt(null, common));
+ }
+
+ // FlinkTopologyContext(StormTopology topology, Map stormConf, Map<Integer, String> taskToComponent,
+ // Map<String, List<Integer>> componentToSortedTasks, Map<String, Map<String, Fields>>
+ // componentToStreamToFields, String stormId, String codeDir, String pidDir, Integer taskId, Integer
+ // workerPort, List<Integer> workerTasks, Map<String, Object> defaultResources, Map<String, Object>
+ // userResources, Map<String, Object> executorData, Map registeredMetrics, Atom openOrPrepareWasCalled)
+
+ return new FlinkTopologyContext(new StormTopology(spoutSpecs, bolts, null), null, taskToComponents, null, null,
+ null, null, null, taskId, null, null, null, null, null, null, null);
+ }
--- End diff --
TopologyContext is not supported so far. This is only a pre-alpha version to make it compile and run. Added TopologyContext (in a nice way) need some more work. Not sure If we need it for the first pull request?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27810581
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopology.java ---
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import backtype.storm.generated.StormTopology;
+
+
+
+
+
+/**
+ * {@link FlinkTopology} mimics a {@link StormTopology} and is implemented in terms of a
+ * {@link StreamExecutionEnvironment} . In contrast to a regular {@link StreamExecutionEnvironment}, a
+ * {@link FlinkTopology} cannot be executed directly, but must be handed over to a {@link FlinkLocalCluster},
+ * {@link FlinkSubmitter}, or {@link FlinkClient}.
+ *
+ * @author mjsax
+ */
+class FlinkTopology extends StreamExecutionEnvironment {
+ /**
+ * The corresponding {@link StormTopology} that is mimiced by this {@link FlinkTopology}.
+ */
+ private final StormTopology stormTopology;
+ /**
+ * The number of declared tasks for the whole program (ie, sum over all dops)
+ */
+ private int numberOfTasks = 0;
+
+
+
+ /**
+ * Instantiates a new {@link FlinkTestTopology}.
+ */
+ public FlinkTopology(final StormTopology stormTopology) {
+ super.setParallelism(1);
+ this.stormTopology = stormTopology;
+ }
+
--- End diff --
Storm uses default parallelism of one, thus, we set it to one here. We do not want to break potential user assumptions about default dop.
It also works for different dop values; the user has to specify non-default dop when adding Spouts/Bolts to the topology.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27830643
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopology.java ---
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import backtype.storm.generated.StormTopology;
+
+
+
+
+
+/**
+ * {@link FlinkTopology} mimics a {@link StormTopology} and is implemented in terms of a
+ * {@link StreamExecutionEnvironment} . In contrast to a regular {@link StreamExecutionEnvironment}, a
+ * {@link FlinkTopology} cannot be executed directly, but must be handed over to a {@link FlinkLocalCluster},
+ * {@link FlinkSubmitter}, or {@link FlinkClient}.
+ *
+ * @author mjsax
+ */
+class FlinkTopology extends StreamExecutionEnvironment {
+ /**
+ * The corresponding {@link StormTopology} that is mimiced by this {@link FlinkTopology}.
+ */
+ private final StormTopology stormTopology;
+ /**
+ * The number of declared tasks for the whole program (ie, sum over all dops)
+ */
+ private int numberOfTasks = 0;
+
+
+
+ /**
+ * Instantiates a new {@link FlinkTestTopology}.
+ */
+ public FlinkTopology(final StormTopology stormTopology) {
+ super.setParallelism(1);
+ this.stormTopology = stormTopology;
+ }
+
--- End diff --
Ok, cool. Then just let us put a comment there, that this value is to mirror Storm behaviour.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808440
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkOutputFieldsDeclarer.java ---
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.util.List;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkOutputFieldsDeclarer} is used to get the declared output schema of a {@link IRichSpout spout} or
+ * {@link IRichBolt bolt}.<br />
+ * <br />
+ * <strong>CAUTION: Currently, Flink does only support the default output stream. Furthermore, direct emit is not
+ * supported.</strong>
+ *
+ * @author mjsax
+ */
+final class FlinkOutputFieldsDeclarer implements OutputFieldsDeclarer {
+ private Fields outputSchema;
+
+ @Override
+ public void declare(final Fields fields) {
+ this.declareStream(Utils.DEFAULT_STREAM_ID, false, fields);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * Direct emit is no supported by Flink. Parameter {@code direct} must be {@code false}.
+ *
+ * @throws UnsupportedOperationException
+ * if {@code direct} is {@code true}
+ */
+ @Override
+ public void declare(final boolean direct, final Fields fields) {
+ this.declareStream(Utils.DEFAULT_STREAM_ID, direct, fields);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * Currently, Flink only supports the default output stream. Thus, pareamter {@code streamId} must be equals to
+ * {@link Utils#DEFAULT_STREAM_ID}.
+ *
+ * @throws UnsupportedOperationException
+ * if {@code streamId} is not equal to {@link Utils#DEFAULT_STREAM_ID}
+ */
+ @Override
+ public void declareStream(final String streamId, final Fields fields) {
+ this.declareStream(streamId, false, fields);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * Currently, Flink only supports the default output stream. Thus, pareamter {@code streamId} must be equals to
+ * {@link Utils#DEFAULT_STREAM_ID}. Furthermore, direct emit is no supported by Flink and parameter {@code direct}
+ * must be {@code false}.
+ *
+ * @throws UnsupportedOperationException
+ * if {@code streamId} is not equal to {@link Utils#DEFAULT_STREAM_ID} or {@code direct} is {@code true}
+ */
+ @Override
+ public void declareStream(final String streamId, final boolean direct, final Fields fields) {
+ if(!Utils.DEFAULT_STREAM_ID.equals(streamId)) {
+ throw new UnsupportedOperationException("Currently, only the default output stream is supported by Flink");
+ }
+ if(direct) {
+ throw new UnsupportedOperationException("Direct emit is not supported by Flink");
+ }
+
+ this.outputSchema = fields;
+ }
+
+ /**
+ * Returns {@link TypeInformation} for the declared output schema. If no or an empty output schema was declared,
+ * {@code null} is returned.
+ *
+ * @return output type information for the declared output schema; or {@code null} if no output schema was declared
+ *
+ * @throws IllegalArgumentException
+ * if more then 25 attributes are declared
+ */
+ public TypeInformation<?> getOutputType() throws IllegalArgumentException {
+ if((this.outputSchema == null) || (this.outputSchema.size() == 0)) {
+ return null;
+ }
+
+ Tuple t;
+ final int numberOfAttributes = this.outputSchema.size();
+
+ if(numberOfAttributes == 1) {
+ return TypeExtractor.getForClass(Object.class);
+ } else if(numberOfAttributes <= 25) {
+ try {
+ t = Tuple.getTupleClass(numberOfAttributes).newInstance();
+ } catch(final InstantiationException e) {
+ throw new RuntimeException(e);
+ } catch(final IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ throw new IllegalArgumentException("Flink supports only a maximum number of 25 attributes.");
+ }
+
+ // TODO: declare only key fields as DummyComparable
+ for(int i = 0; i < numberOfAttributes; ++i) {
+ t.setField(new DummyComparable(), i);
+ }
+
+ return TypeExtractor.getForObject(t);
+ }
+
+ /**
+ * {@link DummyComparable} is a {@link Comparable} helper class that is used to get the correct
+ * {@link TypeInformation} from {@link TypeExtractor} within {@link #getOutputType()}. If key fields are not
+ * comparable, Flink cannot use them and will throw an exception.
+ *
--- End diff --
A not word is missing from the javadoc. If keys are not comparable then an exception is thrown. Maybe a rename would also be nice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27809113
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelper.java ---
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.wrappers;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.flink.api.java.tuple.Tuple25;
+import org.apache.flink.stormcompatibility.api.FlinkTopologyContext;
+import org.apache.flink.streaming.api.streamvertex.StreamingRuntimeContext;
+
+import backtype.storm.generated.Bolt;
+import backtype.storm.generated.ComponentCommon;
+import backtype.storm.generated.SpoutSpec;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.IComponent;
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.IRichSpout;
+
+
+
+
+
+/**
+ * {@link StormWrapperSetupHelper} is an helper class used by {@link AbstractStormSpoutWrapper} or
+ * {@link StormBoltWrapper}.
+ *
+ * @author mjsax
+ */
+class StormWrapperSetupHelper {
+
+ /**
+ * Computes the number of output attributes used by a {@link AbstractStormSpoutWrapper} or {@link StormBoltWrapper}.
+ * Returns zero for raw output type or a value within range [1;25] for output type {@link Tuple1} to {@link Tuple25}
+ * . In case of a data sink, {@code -1} is returned. .
+ *
+ * @param spoutOrBolt
+ * The Storm {@link IRichSpout spout} or {@link IRichBolt bolt} to be used.
+ * @param rawOutput
+ * Set to {@code true} if a single attribute output stream, should not be of type {@link Tuple1} but be
+ * of a raw type.
+ * @return The number of attributes to be used.
+ * @throws IllegalArgumentException
+ * If {@code rawOuput} is {@code true} and the number of declared output attributes is not 1 or if
+ * {@code rawOuput} is {@code false} and the number of declared output attributes is not with range
+ * [1;25].
+ */
+ public static int getNumberOfAttributes(final IComponent spoutOrBolt, final boolean rawOutput)
+ throws IllegalArgumentException {
+ final StormOutputFieldsDeclarer declarer = new StormOutputFieldsDeclarer();
+ spoutOrBolt.declareOutputFields(declarer);
+
+ final int declaredNumberOfAttributes = declarer.getNumberOfAttributes();
+
+ if(declaredNumberOfAttributes == -1) {
+ return -1;
+ }
+
+ if((declaredNumberOfAttributes < 1) || (declaredNumberOfAttributes > 25)) {
+ throw new IllegalArgumentException(
+ "Provided bolt declares non supported number of output attributes. Must be in range [1;25] but was "
+ + declaredNumberOfAttributes);
+ }
+
+ if(rawOutput) {
+ if(declaredNumberOfAttributes > 1) {
+ throw new IllegalArgumentException(
+ "Ouput type is requested to be raw type, but provided bolt declares more then one output attribute.");
+
+ }
+ return 0;
+ }
+
+ return declaredNumberOfAttributes;
+ }
+
+ /**
+ * TODO
+ */
+ public static TopologyContext convertToTopologyContext(final StreamingRuntimeContext context, final boolean spoutOrBolt) {
+ final Integer taskId = new Integer(1 + context.getIndexOfThisSubtask());
+
+ final Map<Integer, String> taskToComponents = new HashMap<Integer, String>();
+ taskToComponents.put(taskId, context.getTaskName());
+
+ final ComponentCommon common = new ComponentCommon();
+ common.set_parallelism_hint(context.getNumberOfParallelSubtasks());
+
+ final Map<String, Bolt> bolts = new HashMap<String, Bolt>();
+ final Map<String, SpoutSpec> spoutSpecs = new HashMap<String, SpoutSpec>();
+
+ if(spoutOrBolt) {
+ spoutSpecs.put(context.getTaskName(), new SpoutSpec(null, common));
+ } else {
+ bolts.put(context.getTaskName(), new Bolt(null, common));
+ }
+
+ // FlinkTopologyContext(StormTopology topology, Map stormConf, Map<Integer, String> taskToComponent,
+ // Map<String, List<Integer>> componentToSortedTasks, Map<String, Map<String, Fields>>
+ // componentToStreamToFields, String stormId, String codeDir, String pidDir, Integer taskId, Integer
+ // workerPort, List<Integer> workerTasks, Map<String, Object> defaultResources, Map<String, Object>
+ // userResources, Map<String, Object> executorData, Map registeredMetrics, Atom openOrPrepareWasCalled)
+
+ return new FlinkTopologyContext(new StormTopology(spoutSpecs, bolts, null), null, taskToComponents, null, null,
+ null, null, null, taskId, null, null, null, null, null, null, null);
+ }
--- End diff --
That constructor needs a revision. :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30112988
--- Diff: flink-staging/flink-streaming/flink-storm-examples/src/main/java/org/apache/flink/stormcompatibility/wordcount/stormoperators/StormFileSpout.java ---
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.wordcount.stormoperators;
+
+import java.io.BufferedReader;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.Map;
+
+import backtype.storm.spout.SpoutOutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.tuple.Values;
+
+
+
+
+
+/**
+ * Implements a Storm Spout that reads data from a given local file.
+ */
+public final class StormFileSpout extends AbstractStormSpout {
+ private static final long serialVersionUID = -6996907090003590436L;
+
+ private final String path;
+ private BufferedReader reader;
+
+
+
+ public StormFileSpout(final String path) {
+ this.path = path;
+ }
+
+
+
+ @Override
+ public void open(@SuppressWarnings("rawtypes") final Map conf, final TopologyContext context, @SuppressWarnings("hiding") final SpoutOutputCollector collector) {
+ super.open(conf, context, collector);
+ try {
+ this.reader = new BufferedReader(new FileReader(this.path));
+ } catch(final FileNotFoundException e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ public void close() {
+ if(this.reader != null) {
+ try {
+ this.reader.close();
+ } catch(final IOException e) {
+ e.printStackTrace();
--- End diff --
These exceptions should be re-thrown as a runtime exception so that the system can properly fail (instead of logging a stacktrace to sysout
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-112023822
Nice :)
Why did you turn off logging? I have some randomized tests and if they fail, the test error cannot be reproduces without the seed value that goes to the log...
About canceling jobs: I have already an open pull request for it: https://github.com/apache/flink/pull/750
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30113838
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+import org.apache.flink.runtime.messages.JobManagerMessages.RunningJobsStatus;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.Nimbus;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link Nimbus}{@code .Client} at once, to interact with
+ * Flink's JobManager instead of Storm's Nimbus.
+ */
+public class FlinkClient {
+ /**
+ * The jobmanager's host name.
+ */
+ private final String jobManagerHost;
+ /**
+ * The jobmanager's rpc port.
+ */
+ private final int jobManagerPort;
+ /**
+ * The user specified timeout in milliseconds.
+ */
+ private final String timeout;
+
+
+
+ /*
+ * The following methods are derived from "backtype.storm.utils.NimbusClient"
+ */
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port) {
+ this(conf, host, port, null);
+ }
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ * @param timeout
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port, final Integer timeout) {
--- End diff --
The `conf` argument is unsued
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808488
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkSubmitter.java ---
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.Nimbus;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkSubmitter} mimics a {@link StromSubmitter} to submit Storm topologies to a Flink cluster.
+ *
+ * @author mjsax
+ */
+public class FlinkSubmitter {
+ public static Logger logger = LoggerFactory.getLogger(FlinkSubmitter.class);
+
+ private static Nimbus.Iface localNimbus = null;
+
+ public static void setLocalNimbus(final Nimbus.Iface localNimbusHandler) {
+ FlinkSubmitter.localNimbus = localNimbusHandler;
+ }
+
+ /**
+ * Submits a topology to run on the cluster. A topology runs forever or until explicitly killed.
+ *
+ *
+ * @param name
+ * the name of the storm.
+ * @param stormConf
+ * the topology-specific configuration. See {@link Config}.
+ * @param topology
+ * the processing to execute.
+ * @throws AlreadyAliveException
+ * if a topology with this name is already running
+ * @throws InvalidTopologyException
+ * if an invalid topology was submitted
+ */
+ public static void submitTopology(final String name, final Map<?, ?> stormConf, final FlinkTopology topology)
+ throws AlreadyAliveException, InvalidTopologyException {
+ submitTopology(name, stormConf, topology, (SubmitOptions)null, (FlinkProgressListener)null);
+ }
+
+ /**
+ * Submits a topology to run on the cluster. A topology runs forever or until explicitly killed.
+ *
+ * @param name
+ * the name of the storm.
+ * @param stormConf
+ * the topology-specific configuration. See {@link Config}.
+ * @param topology
+ * the processing to execute.
+ * @param opts
+ * to manipulate the starting of the topology.
+ * @throws AlreadyAliveException
+ * if a topology with this name is already running
+ * @throws InvalidTopologyException
+ * if an invalid topology was submitted
+ */
+ public static void submitTopology(final String name, final Map<?, ?> stormConf, final FlinkTopology topology, final SubmitOptions opts)
+ throws AlreadyAliveException, InvalidTopologyException {
+ submitTopology(name, stormConf, topology, opts, (FlinkProgressListener)null);
+ }
+
+ /**
+ * Submits a topology to run on the cluster. A topology runs forever or until explicitly killed. The given
+ * {@link FlinkProgressListener} is ignored because progress bars are not supported by Flink.
+ *
+ *
+ * @param name
+ * the name of the storm.
+ * @param stormConf
+ * the topology-specific configuration. See {@link Config}.
+ * @param topology
+ * the processing to execute.
+ * @param opts
+ * to manipulate the starting of the topology
+ * @param progressListener
+ * to track the progress of the jar upload process
+ * @throws AlreadyAliveException
+ * if a topology with this name is already running
+ * @throws InvalidTopologyException
+ * if an invalid topology was submitted
+ */
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public static void submitTopology(final String name, final Map stormConf, final FlinkTopology topology, final SubmitOptions opts, final FlinkProgressListener progressListener)
+ throws AlreadyAliveException, InvalidTopologyException {
+ if(!Utils.isValidConf(stormConf)) {
+ throw new IllegalArgumentException("Storm conf is not valid. Must be json-serializable");
+ }
+
+ final HashMap localStormConf = new HashMap(stormConf);
+ localStormConf.putAll(Utils.readCommandLineOpts());
+
+ final Map conf = Utils.readStormConfig();
+ conf.putAll(localStormConf);
+
+ final String serConf = JSONValue.toJSONString(localStormConf);
+
+ if(localNimbus != null) {
+ // TODO: check what is the difference about localNimbus vs LocalCluster
+ // LOG.info("Submitting topology " + name + " in local mode");
--- End diff --
Let us resolve this, before merging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30117273
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelperTest.java ---
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.wrappers;
+
+import static org.mockito.Mockito.mock;
+
+import java.util.Random;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import backtype.storm.topology.IComponent;
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.tuple.Fields;
+
+
+
+
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(StormWrapperSetupHelper.class)
+public class StormWrapperSetupHelperTest {
+ private Random r;
+
+
+
+ @Before
+ public void prepareTest() {
+ final long seed = System.currentTimeMillis();
+ this.r = new Random(seed);
+ System.out.println("Test seed: " + seed);
--- End diff --
Sure. What logging level should I use? I want to make sure to see the seed anytime a build fails to be able to reproduce the test.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30113138
--- Diff: flink-staging/flink-streaming/flink-storm-examples/src/main/java/org/apache/flink/stormcompatibility/wordcount/stormoperators/StormBoltFileSink.java ---
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.wordcount.stormoperators;
+
+import java.io.BufferedWriter;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Map;
+
+import backtype.storm.task.TopologyContext;
+
+
+
+
+
+/**
+ * Implements a sink that write the received data to the given file (as a result of {@code Object.toString()} for each
+ * attribute).
+ */
+public final class StormBoltFileSink extends AbstractStormBoltSink {
+ private static final long serialVersionUID = 2014027288631273666L;
+
+ private final String path;
+ private BufferedWriter writer;
+
+
+
+ public StormBoltFileSink(final String path) {
+ this.path = path;
+ }
+
+
+
+ @Override
+ public void prepareSimple(@SuppressWarnings("rawtypes") final Map stormConf, final TopologyContext context) {
+ try {
+ this.writer = new BufferedWriter(new FileWriter(this.path));
+ } catch(final IOException e) {
+ e.printStackTrace();
--- End diff --
Again, the exception handling in this file is not correct
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-111990358
Turned logging off for tests by default and did minor javadoc and typo cleanup. We should address the job cancelling throwing expected exceptions soonish. But I think we can merge it to contrib in the current state.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-111878533
Did a final pass and a rebase, code now looks good. There is one little issue with the tests that I need to check again tomorrow morning then I'm merging it. :+1:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-110540528
Hi, I rebased to the current master (and new streaming API) as discussed and cleaned up everything. Unfortunately, I was not able to cherry-pick the last commits:
- @szape commits: https://github.com/mbalassi/flink/commit/ed189aac5de79f1af50efcbb0a1d9defcc582d1a
- @mbalassi https://github.com/mbalassi/flink/commit/9704649267dd29291bb14f223f17a18fc2c26cd0
- @mbalassi https://github.com/mbalassi/flink/commit/3837880544149361a74a2b9b915346e41c46dbac
@szape commit is nor crucial and can be applied right now (please per pull request to mjsax/flink-storm-compatibiltiy) or later on.
Both commits from @mbalassi are critical and I re-did the changes manually (see last commit of this PR). Of course, I accounted it to @mbalassi as an author.
From my point of view, this PR is ready to be merged (hope that Travis will be green).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808567
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopology.java ---
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import backtype.storm.generated.StormTopology;
+
+
+
+
+
+/**
+ * {@link FlinkTopology} mimics a {@link StormTopology} and is implemented in terms of a
+ * {@link StreamExecutionEnvironment} . In contrast to a regular {@link StreamExecutionEnvironment}, a
+ * {@link FlinkTopology} cannot be executed directly, but must be handed over to a {@link FlinkLocalCluster},
+ * {@link FlinkSubmitter}, or {@link FlinkClient}.
+ *
+ * @author mjsax
+ */
+class FlinkTopology extends StreamExecutionEnvironment {
+ /**
+ * The corresponding {@link StormTopology} that is mimiced by this {@link FlinkTopology}.
+ */
+ private final StormTopology stormTopology;
+ /**
+ * The number of declared tasks for the whole program (ie, sum over all dops)
+ */
+ private int numberOfTasks = 0;
+
+
+
+ /**
+ * Instantiates a new {@link FlinkTestTopology}.
+ */
+ public FlinkTopology(final StormTopology stormTopology) {
+ super.setParallelism(1);
+ this.stormTopology = stormTopology;
+ }
+
--- End diff --
Does it work for higher parallelism?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27822552
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.ClusterSummary;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link backtype.storm.generated.Nimbus.Client
+ * Nimbus.Client} at once, to interact with Flink's JobManager instead of Storm's Nimbus.
+ *
+ * @author mjsax
+ */
+public class FlinkClient {
+ /**
+ * Configuration key to specify the location of the file storm-core-0.9.4.jar (for org.apache.storm).
+ */
+ public final static String FLINK_STROM_CORE_JAR_LOCATION = "flink.storm.jar-location.storm-core";
+ /**
+ * Configuration key to specify the location of the file json-simple-1.1.jar (for com.googlecode.json-simple).
+ */
+ public final static String FLINK_STROM_JSON_SIMPLE_JAR_LOCATION = "flink.storm.jar-location.json-simple";
+
+
+
+ /**
+ * The users specified configuration.
+ */
+ private final Map<?, ?> config;
+ /**
+ * The jobmanager's host name.
+ */
+ private final String jobManagerHost;
+ /**
+ * The jobmanager's rpc port.
+ */
+ private final int jobManagerPort;
+ /**
+ * The user specified timeout in milliseconds.
+ */
+ private final String timeout;
+ /**
+ * Maps the names of running topologies to the corresponding Flink {@link JobID}.
+ */
+ private final Map<String, JobID> topologyNameToJobId = new HashMap<String, JobID>();
+ /**
+ * Maps the names of running topologies to the corresponding {@link FlinkTopology}.
+ */
+ private final Map<String, FlinkTopology> topologies = new HashMap<String, FlinkTopology>();
+
+
+
+ /*
+ * The following methods are derived from "backtype.storm.utils.NimbusClient"
+ */
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port) {
+ this(conf, host, port, null);
+ }
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ * @param timeout
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port, final Integer timeout) {
+ this.config = conf;
+ this.jobManagerHost = host;
+ this.jobManagerPort = port;
+ if(timeout != null) {
+ this.timeout = timeout + " ms";
+ } else {
+ this.timeout = null;
+ }
+ }
+
+
+
+ /**
+ * Returns a {@link FlinkClient} that uses the configured {@link Config#NIMBUS_HOST} and
+ * {@link Config#NIMBUS_THRIFT_PORT} as JobManager address.
+ *
+ * @param conf
+ * Configuration that contains the jobmanager's hostname and port.
+ *
+ * @return A configured {@link FlinkClient}.
+ */
+ public static FlinkClient getConfiguredClient(@SuppressWarnings("rawtypes") final Map conf) {
+ final String nimbusHost = (String)conf.get(Config.NIMBUS_HOST);
+ final int nimbusPort = Utils.getInt(conf.get(Config.NIMBUS_THRIFT_PORT)).intValue();
+ return new FlinkClient(conf, nimbusHost, nimbusPort);
+ }
+
+
+
+ /**
+ * Return a reference to itself.
+ *
+ * {@link FlinkClient} mimics both, {@link NimbusClient} and {@link backtype.storm.generated.Nimubs.Client
+ * Nimbus.Client}, at once.
+ *
+ * @return A reference to itself.
+ */
+ public FlinkClient getClient() {
+ return this;
+ }
+
+ public void close() { /* nothing to do */}
+
+
+ /*
+ * The following methods are derived from "backtype.storm.generated.Nimubs.Client"
+ */
+
+ /**
+ * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
+ * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
+ */
+ public void submitTopology(final String name, final String uploadedJarLocation, final String jsonConf, final FlinkTopology topology)
+ throws AlreadyAliveException, InvalidTopologyException {
+ this.submitTopologyWithOpts(name, uploadedJarLocation, jsonConf, topology, null);
+ }
+
+ /**
+ * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
+ * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
+ */
+ @SuppressWarnings("unused")
+ public void submitTopologyWithOpts(final String name, final String uploadedJarLocation, final String jsonConf, final FlinkTopology topology, final SubmitOptions options)
+ throws AlreadyAliveException, InvalidTopologyException {
+
+ // TODO improve by checking job manager, if running
+ if(this.topologyNameToJobId.containsKey(name)) {
+ throw new AlreadyAliveException();
+ }
+
+ // TODO refactor and combine shared code (copied from RemoveStreamEnvironment)
+ try {
+ final JobGraph jobGraph = topology.getStreamGraph().getJobGraph(name);
+
+ // TODO improve error handling for missing config values and uploadedJarLocation==null
+ final String stormCoreJar = (String)this.config.get(FLINK_STROM_CORE_JAR_LOCATION);
+ final String jsonSimpleJar = (String)this.config.get(FLINK_STROM_JSON_SIMPLE_JAR_LOCATION);
+ final String[] jarFileNames = new String[] {stormCoreJar, jsonSimpleJar, uploadedJarLocation};
+
+ final List<File> jarFiles = new ArrayList<File>();
+ for(int i = 0; i < jarFileNames.length; i++) {
+ final File file = new File(jarFileNames[i]);
+ try {
+ JobWithJars.checkJarFile(file);
+ } catch(final IOException e) {
+ throw new RuntimeException("Problem with jar file " + file.getAbsolutePath(), e);
+ }
+ jarFiles.add(file);
+ }
+
+ for(final File file : jarFiles) {
+ jobGraph.addJar(new Path(file.getAbsolutePath()));
+ }
+
+ final Configuration configuration = jobGraph.getJobConfiguration();
+ final Client client = new Client(new InetSocketAddress(this.jobManagerHost, this.jobManagerPort),
+ configuration, JobWithJars.buildUserCodeClassLoader(jarFiles, JobWithJars.class.getClassLoader()), -1);
+
+ try {
+ final JobSubmissionResult result = client.run(jobGraph, false);
+ this.topologyNameToJobId.put(name, result.getJobID());
+ this.topologies.put(name, topology);
+ } catch(final ProgramInvocationException e) {
+ throw new RuntimeException("Cannot execute job due to ProgramInvocationException", e);
+ }
+
+ } catch(final Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ public void killTopology(final String name) throws NotAliveException {
+ this.killTopologyWithOpts(name, null);
+ }
+
+ @SuppressWarnings("unused")
+ public void killTopologyWithOpts(final String name, final KillOptions options) throws NotAliveException {
+ final Configuration configuration = GlobalConfiguration.getConfiguration();
+ if(this.timeout != null) {
+ configuration.setString(ConfigConstants.AKKA_ASK_TIMEOUT, this.timeout);
+ }
+ final FiniteDuration askTimeout = AkkaUtils.getTimeout(configuration);
+
+ ActorSystem actorSystem;
+ try {
+ final scala.Tuple2<String, Object> systemEndpoint = new scala.Tuple2<String, Object>("", new Integer(0));
+ actorSystem = AkkaUtils.createActorSystem(configuration, new Some<scala.Tuple2<String, Object>>(
+ systemEndpoint));
+ } catch(final Exception e) {
+ throw new RuntimeException("Could not start actor system to communicate with JobManager", e);
+ }
+
+ ActorRef jobManager;
+ try {
+ jobManager = JobManager.getJobManagerRemoteReference(new InetSocketAddress(this.jobManagerHost,
+ this.jobManagerPort), actorSystem, AkkaUtils.getLookupTimeout(configuration));
+
+ if(options != null) {
+ try {
+ Thread.sleep(1000 * options.get_wait_secs());
+ } catch(final InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+
+ final Future<Object> response = Patterns.ask(jobManager, new CancelJob(this.topologyNameToJobId.get(name)),
+ new Timeout(askTimeout));
+
+ try {
+ Await.result(response, askTimeout);
+ } catch(final Exception e) {
+ throw new RuntimeException("Killing topology " + name + " with Flink job ID "
+ + this.topologyNameToJobId.get(name) + " failed.", e);
+ }
+
+ this.topologyNameToJobId.remove(name);
+ this.topologies.remove(name);
+ } catch(final IOException e) {
+ throw new RuntimeException("Could not connect to Flink JobManager with address " + this.jobManagerHost
+ + ":" + this.jobManagerPort, e);
+ }
+ }
+
+ // not sure if the following methods can be supported or not...
+
+ // public void activate(final String name) throws NotAliveException {}
+
+ // public void deactivate(final String name) throws NotAliveException {}
+
+ // public void rebalance(final String name, final RebalanceOptions options) throws NotAliveException,
+ // InvalidTopologyException {}
+
+ // public String beginFileUpload() {
+ // return null;
+ // }
+
+ // public void uploadChunk(final String location, final ByteBuffer chunk) {}
+
+ // public void finishFileUpload(final String location) {}
+
+ // public String beginFileDownload(final String file) {
+ // return null;
+ // }
+
+ // public ByteBuffer downloadChunk(final String id) {
+ // return null;
+ // }
+
+ // public void send_downloadChunk(final String id) {}
+
+ // public String getNimbusConf() {
+ // return null;
+ // }
--- End diff --
I would like to keep it or to change it to throw UnsupportedOperationException(). This ensures, that the user is aware what operations are supported (or not supported) by the compability layer.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808721
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyBuilder.java ---
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.stormcompatibility.wrappers.StormBoltWrapper;
+import org.apache.flink.stormcompatibility.wrappers.StormSpoutWrapper;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+import backtype.storm.generated.ComponentCommon;
+import backtype.storm.generated.GlobalStreamId;
+import backtype.storm.generated.Grouping;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.topology.BasicBoltExecutor;
+import backtype.storm.topology.BoltDeclarer;
+import backtype.storm.topology.IBasicBolt;
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.topology.SpoutDeclarer;
+import backtype.storm.topology.TopologyBuilder;
+
+
+
+
+
+/**
+ * {@link FlinkTopologyBuilder} mimics a {@link TopologyBuilder}, but builds a Flink program instead of a Storm
+ * topology. Most methods (except {@link #createTopology()} are copied from the original {@link TopologyBuilder}
+ * implementation to ensure equal behavior.<br />
+ * <br />
+ * <strong>CAUTION: {@link StateSpout}s and multiple output streams per spout/bolt are currently not supported.</strong>
+ *
+ * @author mjsax
+ */
+public class FlinkTopologyBuilder {
+ /**
+ * A Storm {@link TopologyBuilder} to build a real Storm topology.
+ */
+ private final TopologyBuilder stormBuilder = new TopologyBuilder();
+ /**
+ * All user spouts by their ID.
+ */
+ private final HashMap<String, IRichSpout> spouts = new HashMap<String, IRichSpout>();
+ /**
+ * All user bolts by their ID.
+ */
+ private final HashMap<String, IRichBolt> bolts = new HashMap<String, IRichBolt>();
+
+ // TODO add StateSpout support (Storm 0.9.4 does not yet support StateSpouts itself)
+
+
+
+ /**
+ * Creates a Flink program that used the specified spouts and bolts.
+ */
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public FlinkTopology createTopology() {
+ final StormTopology stormTopolgoy = this.stormBuilder.createTopology();
--- End diff --
Nice! :+1:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808253
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.ClusterSummary;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link backtype.storm.generated.Nimbus.Client
+ * Nimbus.Client} at once, to interact with Flink's JobManager instead of Storm's Nimbus.
+ *
+ * @author mjsax
+ */
+public class FlinkClient {
+ /**
+ * Configuration key to specify the location of the file storm-core-0.9.4.jar (for org.apache.storm).
+ */
+ public final static String FLINK_STROM_CORE_JAR_LOCATION = "flink.storm.jar-location.storm-core";
+ /**
+ * Configuration key to specify the location of the file json-simple-1.1.jar (for com.googlecode.json-simple).
+ */
+ public final static String FLINK_STROM_JSON_SIMPLE_JAR_LOCATION = "flink.storm.jar-location.json-simple";
+
+
+
+ /**
+ * The users specified configuration.
+ */
+ private final Map<?, ?> config;
+ /**
+ * The jobmanager's host name.
+ */
+ private final String jobManagerHost;
+ /**
+ * The jobmanager's rpc port.
+ */
+ private final int jobManagerPort;
+ /**
+ * The user specified timeout in milliseconds.
+ */
+ private final String timeout;
+ /**
+ * Maps the names of running topologies to the corresponding Flink {@link JobID}.
+ */
+ private final Map<String, JobID> topologyNameToJobId = new HashMap<String, JobID>();
+ /**
+ * Maps the names of running topologies to the corresponding {@link FlinkTopology}.
+ */
+ private final Map<String, FlinkTopology> topologies = new HashMap<String, FlinkTopology>();
+
+
+
+ /*
+ * The following methods are derived from "backtype.storm.utils.NimbusClient"
+ */
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port) {
+ this(conf, host, port, null);
+ }
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ * @param timeout
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port, final Integer timeout) {
+ this.config = conf;
+ this.jobManagerHost = host;
+ this.jobManagerPort = port;
+ if(timeout != null) {
+ this.timeout = timeout + " ms";
+ } else {
+ this.timeout = null;
+ }
+ }
+
+
+
+ /**
+ * Returns a {@link FlinkClient} that uses the configured {@link Config#NIMBUS_HOST} and
+ * {@link Config#NIMBUS_THRIFT_PORT} as JobManager address.
+ *
+ * @param conf
+ * Configuration that contains the jobmanager's hostname and port.
+ *
+ * @return A configured {@link FlinkClient}.
+ */
+ public static FlinkClient getConfiguredClient(@SuppressWarnings("rawtypes") final Map conf) {
+ final String nimbusHost = (String)conf.get(Config.NIMBUS_HOST);
+ final int nimbusPort = Utils.getInt(conf.get(Config.NIMBUS_THRIFT_PORT)).intValue();
+ return new FlinkClient(conf, nimbusHost, nimbusPort);
+ }
+
+
+
+ /**
+ * Return a reference to itself.
+ *
+ * {@link FlinkClient} mimics both, {@link NimbusClient} and {@link backtype.storm.generated.Nimubs.Client
+ * Nimbus.Client}, at once.
+ *
+ * @return A reference to itself.
+ */
+ public FlinkClient getClient() {
+ return this;
+ }
+
+ public void close() { /* nothing to do */}
+
+
+ /*
+ * The following methods are derived from "backtype.storm.generated.Nimubs.Client"
+ */
+
+ /**
+ * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
+ * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
+ */
+ public void submitTopology(final String name, final String uploadedJarLocation, final String jsonConf, final FlinkTopology topology)
+ throws AlreadyAliveException, InvalidTopologyException {
+ this.submitTopologyWithOpts(name, uploadedJarLocation, jsonConf, topology, null);
+ }
+
+ /**
+ * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
+ * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
+ */
+ @SuppressWarnings("unused")
+ public void submitTopologyWithOpts(final String name, final String uploadedJarLocation, final String jsonConf, final FlinkTopology topology, final SubmitOptions options)
+ throws AlreadyAliveException, InvalidTopologyException {
+
+ // TODO improve by checking job manager, if running
+ if(this.topologyNameToJobId.containsKey(name)) {
+ throw new AlreadyAliveException();
+ }
+
+ // TODO refactor and combine shared code (copied from RemoveStreamEnvironment)
+ try {
+ final JobGraph jobGraph = topology.getStreamGraph().getJobGraph(name);
+
+ // TODO improve error handling for missing config values and uploadedJarLocation==null
+ final String stormCoreJar = (String)this.config.get(FLINK_STROM_CORE_JAR_LOCATION);
+ final String jsonSimpleJar = (String)this.config.get(FLINK_STROM_JSON_SIMPLE_JAR_LOCATION);
+ final String[] jarFileNames = new String[] {stormCoreJar, jsonSimpleJar, uploadedJarLocation};
+
+ final List<File> jarFiles = new ArrayList<File>();
+ for(int i = 0; i < jarFileNames.length; i++) {
+ final File file = new File(jarFileNames[i]);
+ try {
+ JobWithJars.checkJarFile(file);
+ } catch(final IOException e) {
+ throw new RuntimeException("Problem with jar file " + file.getAbsolutePath(), e);
+ }
+ jarFiles.add(file);
+ }
+
+ for(final File file : jarFiles) {
+ jobGraph.addJar(new Path(file.getAbsolutePath()));
+ }
+
+ final Configuration configuration = jobGraph.getJobConfiguration();
+ final Client client = new Client(new InetSocketAddress(this.jobManagerHost, this.jobManagerPort),
+ configuration, JobWithJars.buildUserCodeClassLoader(jarFiles, JobWithJars.class.getClassLoader()), -1);
+
+ try {
+ final JobSubmissionResult result = client.run(jobGraph, false);
+ this.topologyNameToJobId.put(name, result.getJobID());
+ this.topologies.put(name, topology);
+ } catch(final ProgramInvocationException e) {
+ throw new RuntimeException("Cannot execute job due to ProgramInvocationException", e);
+ }
+
+ } catch(final Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ public void killTopology(final String name) throws NotAliveException {
+ this.killTopologyWithOpts(name, null);
+ }
+
+ @SuppressWarnings("unused")
+ public void killTopologyWithOpts(final String name, final KillOptions options) throws NotAliveException {
+ final Configuration configuration = GlobalConfiguration.getConfiguration();
+ if(this.timeout != null) {
+ configuration.setString(ConfigConstants.AKKA_ASK_TIMEOUT, this.timeout);
+ }
+ final FiniteDuration askTimeout = AkkaUtils.getTimeout(configuration);
+
+ ActorSystem actorSystem;
+ try {
+ final scala.Tuple2<String, Object> systemEndpoint = new scala.Tuple2<String, Object>("", new Integer(0));
+ actorSystem = AkkaUtils.createActorSystem(configuration, new Some<scala.Tuple2<String, Object>>(
+ systemEndpoint));
+ } catch(final Exception e) {
+ throw new RuntimeException("Could not start actor system to communicate with JobManager", e);
+ }
+
+ ActorRef jobManager;
+ try {
+ jobManager = JobManager.getJobManagerRemoteReference(new InetSocketAddress(this.jobManagerHost,
+ this.jobManagerPort), actorSystem, AkkaUtils.getLookupTimeout(configuration));
+
+ if(options != null) {
+ try {
+ Thread.sleep(1000 * options.get_wait_secs());
+ } catch(final InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+
+ final Future<Object> response = Patterns.ask(jobManager, new CancelJob(this.topologyNameToJobId.get(name)),
+ new Timeout(askTimeout));
+
+ try {
+ Await.result(response, askTimeout);
+ } catch(final Exception e) {
+ throw new RuntimeException("Killing topology " + name + " with Flink job ID "
+ + this.topologyNameToJobId.get(name) + " failed.", e);
+ }
+
+ this.topologyNameToJobId.remove(name);
+ this.topologies.remove(name);
+ } catch(final IOException e) {
+ throw new RuntimeException("Could not connect to Flink JobManager with address " + this.jobManagerHost
+ + ":" + this.jobManagerPort, e);
+ }
+ }
+
+ // not sure if the following methods can be supported or not...
+
+ // public void activate(final String name) throws NotAliveException {}
+
+ // public void deactivate(final String name) throws NotAliveException {}
+
+ // public void rebalance(final String name, final RebalanceOptions options) throws NotAliveException,
+ // InvalidTopologyException {}
+
+ // public String beginFileUpload() {
+ // return null;
+ // }
+
+ // public void uploadChunk(final String location, final ByteBuffer chunk) {}
+
+ // public void finishFileUpload(final String location) {}
+
+ // public String beginFileDownload(final String file) {
+ // return null;
+ // }
+
+ // public ByteBuffer downloadChunk(final String id) {
+ // return null;
+ // }
+
+ // public void send_downloadChunk(final String id) {}
+
+ // public String getNimbusConf() {
+ // return null;
+ // }
--- End diff --
This is a bit too much comment here to merge, but we should definitely discuss these.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-90209580
I fixed the rebase problem to remove the unnecessary commits. I also squashed the commits you mentioned, removed author tags, and improved commit messages.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-111848332
Hey @mjsax, thanks for the cleanup. I was focused on the release during the week, I'm doing the merge this evening.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808843
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/examples/BoltTokenizerWordCount.java ---
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.examples;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.examples.java.wordcount.util.WordCountData;
+import org.apache.flink.stormcompatibility.examples.stormoperators.StormBoltTokenizer;
+import org.apache.flink.stormcompatibility.wrappers.StormBoltWrapper;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import backtype.storm.topology.IRichBolt;
+
+
+
+
+
+/**
+ * Implements the "WordCount" program that computes a simple word occurrence histogram over text files in a streaming
+ * fashion. The tokenizer step is performed by a Storm {@link IRichBolt bolt}.
+ *
+ * <p>
+ * The input is a plain text file with lines separated by newline characters.
+ *
+ * <p>
+ * Usage: <code>WordCount <text path> <result path></code><br>
+ * If no parameters are provided, the program is run with default data from {@link WordCountData}.
+ *
+ * <p>
+ * This example shows how to:
+ * <ul>
+ * <li>use a Storm bolt within a Flink Streaming program.
+ * </ul>
+ */
+public class BoltTokenizerWordCount {
+
--- End diff --
Nice! :+1:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27809074
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormTuple.java ---
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.wrappers;
+
+import java.util.List;
+
+import backtype.storm.generated.GlobalStreamId;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.MessageId;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+
+
+
+
+/**
+ * {@link StormTuple} converts a Flink tuple of type {@code IN} into a Storm tuple.
+ *
+ * @author mjsax
+ */
+class StormTuple<IN> implements Tuple {
+ /**
--- End diff --
Looking at this file I am really thankful for the Flink Tuple type. Can't we support the operations you marked unsupported?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808315
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkLocalCluster.java ---
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.util.Map;
+
+import org.apache.flink.streaming.util.ClusterUtil;
+
+import backtype.storm.LocalCluster;
+import backtype.storm.generated.ClusterSummary;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.RebalanceOptions;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.generated.TopologyInfo;
+
+
+
+
+
+/**
+ * {@link FlinkLocalCluster} mimics a Storm {@link LocalCluster}.
+ *
+ * @author mjsax
+ */
+public class FlinkLocalCluster {
+
--- End diff --
The class looks like a stub itself :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808962
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormSpoutWrapper.java ---
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.wrappers;
+
+import org.apache.flink.api.java.tuple.Tuple25;
+
+import backtype.storm.topology.IRichSpout;
+
+
+
+
+
--- End diff --
You use too much spacing in this file.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30113379
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelperTest.java ---
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.wrappers;
+
+import static org.mockito.Mockito.mock;
+
+import java.util.Random;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import backtype.storm.topology.IComponent;
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.tuple.Fields;
+
+
+
+
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(StormWrapperSetupHelper.class)
+public class StormWrapperSetupHelperTest {
+ private Random r;
+
+
+
+ @Before
+ public void prepareTest() {
+ final long seed = System.currentTimeMillis();
+ this.r = new Random(seed);
+ System.out.println("Test seed: " + seed);
--- End diff --
Can you use our logging framework for these messages?
We set different logging levels for different environments (mvn locally, travis, ide)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r28517070
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.ClusterSummary;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link backtype.storm.generated.Nimbus.Client
+ * Nimbus.Client} at once, to interact with Flink's JobManager instead of Storm's Nimbus.
+ *
+ * @author mjsax
+ */
+public class FlinkClient {
+ /**
+ * Configuration key to specify the location of the file storm-core-0.9.4.jar (for org.apache.storm).
+ */
+ public final static String FLINK_STROM_CORE_JAR_LOCATION = "flink.storm.jar-location.storm-core";
+ /**
+ * Configuration key to specify the location of the file json-simple-1.1.jar (for com.googlecode.json-simple).
+ */
+ public final static String FLINK_STROM_JSON_SIMPLE_JAR_LOCATION = "flink.storm.jar-location.json-simple";
--- End diff --
I fixed this issue. (see flink-dist/pom.xml, FlinkClient.java, and StormWordCountRemote.java)
In order to get a clean module structure, I extracted the examples from flink-storm-compatiblitly into a own module flink-storm-examples.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30113961
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+import org.apache.flink.runtime.messages.JobManagerMessages.RunningJobsStatus;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.Nimbus;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link Nimbus}{@code .Client} at once, to interact with
+ * Flink's JobManager instead of Storm's Nimbus.
+ */
+public class FlinkClient {
+ /**
+ * The jobmanager's host name.
+ */
+ private final String jobManagerHost;
+ /**
+ * The jobmanager's rpc port.
+ */
+ private final int jobManagerPort;
+ /**
+ * The user specified timeout in milliseconds.
+ */
+ private final String timeout;
+
+
+
+ /*
+ * The following methods are derived from "backtype.storm.utils.NimbusClient"
+ */
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port) {
+ this(conf, host, port, null);
+ }
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ * @param timeout
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port, final Integer timeout) {
+ this.jobManagerHost = host;
+ this.jobManagerPort = port;
+ if(timeout != null) {
+ this.timeout = timeout + " ms";
+ } else {
+ this.timeout = null;
+ }
+ }
+
+
+
+ /**
+ * Returns a {@link FlinkClient} that uses the configured {@link Config#NIMBUS_HOST} and
+ * {@link Config#NIMBUS_THRIFT_PORT} as JobManager address.
+ *
+ * @param conf
+ * Configuration that contains the jobmanager's hostname and port.
+ *
+ * @return A configured {@link FlinkClient}.
+ */
+ public static FlinkClient getConfiguredClient(@SuppressWarnings("rawtypes") final Map conf) {
+ final String nimbusHost = (String)conf.get(Config.NIMBUS_HOST);
+ final int nimbusPort = Utils.getInt(conf.get(Config.NIMBUS_THRIFT_PORT)).intValue();
+ return new FlinkClient(conf, nimbusHost, nimbusPort);
+ }
+
+
+
+ /**
+ * Return a reference to itself.
+ *
+ * {@link FlinkClient} mimics both, {@link NimbusClient} and {@link Nimbus}{@code .Client}, at once.
+ *
+ * @return A reference to itself.
+ */
+ public FlinkClient getClient() {
+ return this;
+ }
+
+ public void close() {/* nothing to do */}
+
+
+ /*
+ * The following methods are derived from "backtype.storm.generated.Nimubs.Client"
+ */
+
+ /**
+ * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
+ * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
+ */
+ public void submitTopology(final String name, final String uploadedJarLocation, final String jsonConf, final FlinkTopology topology)
+ throws AlreadyAliveException, InvalidTopologyException {
+ this.submitTopologyWithOpts(name, uploadedJarLocation, jsonConf, topology, null);
+ }
+
+ /**
+ * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
+ * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
+ */
+ public void submitTopologyWithOpts(final String name, final String uploadedJarLocation, final String jsonConf, final FlinkTopology topology, final SubmitOptions options)
+ throws AlreadyAliveException, InvalidTopologyException {
+
+ if(this.getTopologyJobId(name) != null) {
+ throw new AlreadyAliveException();
+ }
+
+ final File uploadedJarFile = new File(uploadedJarLocation);
+ try {
+ JobWithJars.checkJarFile(uploadedJarFile);
+ } catch(final IOException e) {
+ throw new RuntimeException("Problem with jar file " + uploadedJarFile.getAbsolutePath(), e);
+ }
+
+ final List<File> jarFiles = new ArrayList<File>();
+ jarFiles.add(uploadedJarFile);
+
+ final JobGraph jobGraph = topology.getStreamGraph().getJobGraph(name);
+ jobGraph.addJar(new Path(uploadedJarFile.getAbsolutePath()));
+
+ final Configuration configuration = jobGraph.getJobConfiguration();
+
+ final Client client;
+ try {
+ client = new Client(new InetSocketAddress(this.jobManagerHost, this.jobManagerPort), configuration,
+ JobWithJars.buildUserCodeClassLoader(jarFiles, JobWithJars.class.getClassLoader()), -1);
+ } catch(final UnknownHostException e) {
+ throw new RuntimeException("Cannot execute job due to UnknownHostException", e);
+ }
+
+ try {
+ client.run(jobGraph, false);
+ } catch(final ProgramInvocationException e) {
+ throw new RuntimeException("Cannot execute job due to ProgramInvocationException", e);
+ }
+ }
+
+ public void killTopology(final String name) throws NotAliveException {
+ this.killTopologyWithOpts(name, null);
+ }
+
+ public void killTopologyWithOpts(final String name, final KillOptions options) throws NotAliveException {
+ final JobID jobId = this.getTopologyJobId(name);
+ if(jobId == null) {
+ throw new NotAliveException();
+ }
+
+ try {
+ final ActorRef jobManager = this.getJobManager();
+
+ if(options != null) {
+ try {
+ Thread.sleep(1000 * options.get_wait_secs());
+ } catch(final InterruptedException e) {
+ e.printStackTrace();
--- End diff --
rethrow as runtime e.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30113871
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+import org.apache.flink.runtime.messages.JobManagerMessages.RunningJobsStatus;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.Nimbus;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link Nimbus}{@code .Client} at once, to interact with
+ * Flink's JobManager instead of Storm's Nimbus.
+ */
+public class FlinkClient {
+ /**
+ * The jobmanager's host name.
+ */
+ private final String jobManagerHost;
+ /**
+ * The jobmanager's rpc port.
+ */
+ private final int jobManagerPort;
+ /**
+ * The user specified timeout in milliseconds.
+ */
+ private final String timeout;
+
+
+
+ /*
+ * The following methods are derived from "backtype.storm.utils.NimbusClient"
+ */
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port) {
+ this(conf, host, port, null);
+ }
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ * @param timeout
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port, final Integer timeout) {
--- End diff --
Why is the Map not typed to Map<String, String>?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-111755768
Any news on this PR?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-90132428
Hey Matthias,
Thanks for opening the pull request as discussed on the mailing list. Could you please provide some description here also?
Please do another update on your PR to better match our coding guidelines:
1. The first six commits are not needed. Th first one is already merged, than you have 5 huge merge commits, that we are trying to avoid.
2. Some of your commit messages need a bit more polishing, the "added raw output type support" and "added API compatibility classes" do not describe your work when merged to the Flink code. I would also suggest to add a [streaming] tag for your commits.
3. Please squash 605de53 to 1519d36, becuase it is just a minor fix for that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27830743
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormSpoutWrapper.java ---
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.wrappers;
+
+import org.apache.flink.api.java.tuple.Tuple25;
+
+import backtype.storm.topology.IRichSpout;
+
+
+
+
+
--- End diff --
It is just better to have a somewhat unified look of our code. No semantic problem of course.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-101301921
Hi, I addressed all your comments. The latest build was successful, too. Please let me know if there is anything else to improve the pull request.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27810893
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java ---
@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import backtype.storm.generated.ComponentCommon;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.hooks.ITaskHook;
+import backtype.storm.metric.api.CombinedMetric;
+import backtype.storm.metric.api.ICombiner;
+import backtype.storm.metric.api.IMetric;
+import backtype.storm.metric.api.IReducer;
+import backtype.storm.metric.api.ReducedMetric;
+import backtype.storm.state.ISubscribedState;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.tuple.Fields;
+import clojure.lang.Atom;
+
+
+
+
+
+/**
+ * {@link FlinkTopologyContext} is a {@link TopologyContext} that overwrites certain method that are not applicable when
+ * a Storm topology is executed within Flink.
+ *
+ * @author mjsax
+ */
+public class FlinkTopologyContext extends TopologyContext {
+ /**
+ * The topology belonging to this context.
+ */
+ private final StormTopology topology;
+
+
+
+ /**
+ * TODO remove unused parameters...
+ *
+ * @param topology
+ * @param stormConf
+ * @param taskToComponent
+ * @param componentToSortedTasks
+ * @param componentToStreamToFields
+ * @param stormId
+ * @param codeDir
+ * @param pidDir
+ * @param taskId
+ * @param workerPort
+ * @param workerTasks
+ * @param defaultResources
+ * @param userResources
+ * @param executorData
+ * @param registeredMetrics
+ * @param openOrPrepareWasCalled
+ */
+ public FlinkTopologyContext(final StormTopology topology, @SuppressWarnings("rawtypes") final Map stormConf,
+ final Map<Integer, String> taskToComponent, final Map<String, List<Integer>> componentToSortedTasks,
+ final Map<String, Map<String, Fields>> componentToStreamToFields, final String stormId, final String codeDir,
+ final String pidDir, final Integer taskId, final Integer workerPort, final List<Integer> workerTasks,
+ final Map<String, Object> defaultResources, final Map<String, Object> userResources,
+ final Map<String, Object> executorData, @SuppressWarnings("rawtypes") final Map registeredMetrics,
+ final Atom openOrPrepareWasCalled) {
+ super(topology, stormConf, taskToComponent, componentToSortedTasks, componentToStreamToFields, stormId,
+ codeDir, pidDir, taskId, workerPort, workerTasks, defaultResources, userResources, executorData,
+ registeredMetrics, openOrPrepareWasCalled);
+ this.topology = topology;
+ }
+
+
+
+ @Override
+ public ComponentCommon getComponentCommon(final String componentId) {
+ return super.getComponentCommon(componentId);
+ // final Bolt bolt = this.topology.get_bolts().get(componentId);
+ // if(bolt != null) {
+ // return bolt.get_common();
+ // }
+ // final SpoutSpec spoutSpec = this.topology.get_spouts().get(componentId);
--- End diff --
I prefer throwing exceptions, such that the user is aware what Storm features are not supported. However, this class is in an pre-alpha state... It might be necessary to add some more stuff here, to get meta-data handling right (ie, hand an correct TopooogyContext object to Spouts/Bolts within .open()/.prepare().
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r28344339
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkSubmitter.java ---
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.Nimbus;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkSubmitter} mimics a {@link StromSubmitter} to submit Storm topologies to a Flink cluster.
+ *
+ * @author mjsax
+ */
+public class FlinkSubmitter {
+ public static Logger logger = LoggerFactory.getLogger(FlinkSubmitter.class);
+
+ private static Nimbus.Iface localNimbus = null;
+
+ public static void setLocalNimbus(final Nimbus.Iface localNimbusHandler) {
+ FlinkSubmitter.localNimbus = localNimbusHandler;
+ }
+
+ /**
+ * Submits a topology to run on the cluster. A topology runs forever or until explicitly killed.
+ *
+ *
+ * @param name
+ * the name of the storm.
+ * @param stormConf
+ * the topology-specific configuration. See {@link Config}.
+ * @param topology
+ * the processing to execute.
+ * @throws AlreadyAliveException
+ * if a topology with this name is already running
+ * @throws InvalidTopologyException
+ * if an invalid topology was submitted
+ */
+ public static void submitTopology(final String name, final Map<?, ?> stormConf, final FlinkTopology topology)
+ throws AlreadyAliveException, InvalidTopologyException {
+ submitTopology(name, stormConf, topology, (SubmitOptions)null, (FlinkProgressListener)null);
+ }
+
+ /**
+ * Submits a topology to run on the cluster. A topology runs forever or until explicitly killed.
+ *
+ * @param name
+ * the name of the storm.
+ * @param stormConf
+ * the topology-specific configuration. See {@link Config}.
+ * @param topology
+ * the processing to execute.
+ * @param opts
+ * to manipulate the starting of the topology.
+ * @throws AlreadyAliveException
+ * if a topology with this name is already running
+ * @throws InvalidTopologyException
+ * if an invalid topology was submitted
+ */
+ public static void submitTopology(final String name, final Map<?, ?> stormConf, final FlinkTopology topology, final SubmitOptions opts)
+ throws AlreadyAliveException, InvalidTopologyException {
+ submitTopology(name, stormConf, topology, opts, (FlinkProgressListener)null);
+ }
+
+ /**
+ * Submits a topology to run on the cluster. A topology runs forever or until explicitly killed. The given
+ * {@link FlinkProgressListener} is ignored because progress bars are not supported by Flink.
+ *
+ *
+ * @param name
+ * the name of the storm.
+ * @param stormConf
+ * the topology-specific configuration. See {@link Config}.
+ * @param topology
+ * the processing to execute.
+ * @param opts
+ * to manipulate the starting of the topology
+ * @param progressListener
+ * to track the progress of the jar upload process
+ * @throws AlreadyAliveException
+ * if a topology with this name is already running
+ * @throws InvalidTopologyException
+ * if an invalid topology was submitted
+ */
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public static void submitTopology(final String name, final Map stormConf, final FlinkTopology topology, final SubmitOptions opts, final FlinkProgressListener progressListener)
+ throws AlreadyAliveException, InvalidTopologyException {
+ if(!Utils.isValidConf(stormConf)) {
+ throw new IllegalArgumentException("Storm conf is not valid. Must be json-serializable");
+ }
+
+ final HashMap localStormConf = new HashMap(stormConf);
+ localStormConf.putAll(Utils.readCommandLineOpts());
+
+ final Map conf = Utils.readStormConfig();
+ conf.putAll(localStormConf);
+
+ final String serConf = JSONValue.toJSONString(localStormConf);
+
+ if(localNimbus != null) {
+ // TODO: check what is the difference about localNimbus vs LocalCluster
+ // LOG.info("Submitting topology " + name + " in local mode");
--- End diff --
I ask at the dev-storm mailing list and got the following answer:
"LocalNimbus is something that is not really used. It provides a way for the StormSubmitter to be used when submitting a topology to a local mode cluster. No one uses it and in the past it didn't work, not sure if it works now or not. It is probably best to just not use it unless you really need to abstract out local mode from the topology code used to submit a topology.
- Bobby"
We can ignore this case, and I will remove the code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808060
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.ClusterSummary;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link backtype.storm.generated.Nimbus.Client
+ * Nimbus.Client} at once, to interact with Flink's JobManager instead of Storm's Nimbus.
+ *
+ * @author mjsax
+ */
--- End diff --
No author tags, please.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30113510
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormFiniteSpoutWrapperTest.java ---
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.wrappers;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.LinkedList;
+import java.util.Random;
+
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.tuple.Fields;
+
+
+
+
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(StormWrapperSetupHelper.class)
+public class StormFiniteSpoutWrapperTest {
+ private Random r;
+
+
+
+ @Before
+ public void prepareTest() {
+ final long seed = System.currentTimeMillis();
+ this.r = new Random(seed);
+ System.out.println("Test seed: " + seed);
+ }
--- End diff --
maybe it would make sense to create a test base for the storm compat layer to avoid duplicate code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30117019
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+import org.apache.flink.runtime.messages.JobManagerMessages.RunningJobsStatus;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.Nimbus;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link Nimbus}{@code .Client} at once, to interact with
+ * Flink's JobManager instead of Storm's Nimbus.
+ */
+public class FlinkClient {
+ /**
+ * The jobmanager's host name.
+ */
+ private final String jobManagerHost;
+ /**
+ * The jobmanager's rpc port.
+ */
+ private final int jobManagerPort;
+ /**
+ * The user specified timeout in milliseconds.
+ */
+ private final String timeout;
+
+
+
+ /*
+ * The following methods are derived from "backtype.storm.utils.NimbusClient"
+ */
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port) {
+ this(conf, host, port, null);
+ }
+
+ /**
+ * Instantiates a new {@link FlinkClient} for the given configuration, host name, and port. If values for
+ * {@link Config#NIMBUS_HOST} and {@link Config#NIMBUS_THRIFT_PORT} of the given configuration are ignored.
+ *
+ * @param conf
+ * A configuration.
+ * @param host
+ * The jobmanager's host name.
+ * @param port
+ * The jobmanager's rpc port.
+ * @param timeout
+ */
+ public FlinkClient(final Map<?, ?> conf, final String host, final int port, final Integer timeout) {
--- End diff --
FlinkClient offers the unified interface as NimbusClient and Nimbus.Client (in the compatibility layer, it does not make sense to destingues both). Thus, the constructor offers the same parameters (even in not used right now). The same holds for Map -- Storm does not use any generics.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30113453
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormTupleTest.java ---
@@ -0,0 +1,367 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.wrappers;
+
+import java.util.List;
+import java.util.Random;
+
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+
+
+
+
+public class StormTupleTest {
+ private long seed;
+ private Random r;
+
+
+
+ @Before
+ public void prepare() {
+ this.seed = System.currentTimeMillis();
+ this.r = new Random(this.seed);
+ System.out.println("Test seed: " + this.seed);
--- End diff --
use logging framework
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27811027
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormTuple.java ---
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.wrappers;
+
+import java.util.List;
+
+import backtype.storm.generated.GlobalStreamId;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.MessageId;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+
+
+
+
+/**
+ * {@link StormTuple} converts a Flink tuple of type {@code IN} into a Storm tuple.
+ *
+ * @author mjsax
+ */
+class StormTuple<IN> implements Tuple {
+ /**
--- End diff --
Supporting named attributes is possible (and I guess, not too hard). Should it be part in the first pull request? Or can we add it later on?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808106
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.ClusterSummary;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link backtype.storm.generated.Nimbus.Client
+ * Nimbus.Client} at once, to interact with Flink's JobManager instead of Storm's Nimbus.
+ *
+ * @author mjsax
+ */
+public class FlinkClient {
+ /**
+ * Configuration key to specify the location of the file storm-core-0.9.4.jar (for org.apache.storm).
+ */
+ public final static String FLINK_STROM_CORE_JAR_LOCATION = "flink.storm.jar-location.storm-core";
+ /**
+ * Configuration key to specify the location of the file json-simple-1.1.jar (for com.googlecode.json-simple).
+ */
+ public final static String FLINK_STROM_JSON_SIMPLE_JAR_LOCATION = "flink.storm.jar-location.json-simple";
--- End diff --
Do we need to hard code these?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27810138
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkClient.java ---
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.client.program.Client;
+import org.apache.flink.client.program.JobWithJars;
+import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobManager;
+import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
+
+import scala.Some;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import backtype.storm.Config;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.ClusterSummary;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.generated.KillOptions;
+import backtype.storm.generated.NotAliveException;
+import backtype.storm.generated.SubmitOptions;
+import backtype.storm.utils.NimbusClient;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkClient} mimics a Storm {@link NimbusClient} and {@link backtype.storm.generated.Nimbus.Client
+ * Nimbus.Client} at once, to interact with Flink's JobManager instead of Storm's Nimbus.
+ *
+ * @author mjsax
+ */
+public class FlinkClient {
+ /**
+ * Configuration key to specify the location of the file storm-core-0.9.4.jar (for org.apache.storm).
+ */
+ public final static String FLINK_STROM_CORE_JAR_LOCATION = "flink.storm.jar-location.storm-core";
+ /**
+ * Configuration key to specify the location of the file json-simple-1.1.jar (for com.googlecode.json-simple).
+ */
+ public final static String FLINK_STROM_JSON_SIMPLE_JAR_LOCATION = "flink.storm.jar-location.json-simple";
--- End diff --
No really. It was the least intrusive way to make it work....
In order to be able to execute a Storm topology on Flink, both jars are needed. The build configuration could be changed to include both in the final Flink jar. Because I did not want to change the build without feedback, I added both parameter keys to enalbe the user to specifiy the location for both jars manually.
Right now, both parameters are used within StormWordCountRemote (in a ugly hardcoded way). This could also be removed, if both jars would be included in the final Flink jar.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30116720
--- Diff: flink-staging/flink-streaming/flink-storm-examples/src/main/java/org/apache/flink/stormcompatibility/wordcount/StormWordCountLocal.java ---
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.wordcount;
+
+import org.apache.flink.examples.java.wordcount.util.WordCountData;
+import org.apache.flink.stormcompatibility.api.FlinkLocalCluster;
+import org.apache.flink.stormcompatibility.api.FlinkTopologyBuilder;
+
+import backtype.storm.LocalCluster;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * Implements the "WordCount" program that computes a simple word occurrence histogram over text files in a streaming
+ * fashion. The program is constructed as a regular {@link StormTopology} and submitted to Flink for execution in the
+ * same way as to a Storm {@link LocalCluster}.
+ *
+ * This example shows how to run program directly within Java, thus it cannot be used to submit a {@link StormTopology}
+ * via Flink command line clients (ie, bin/flink).
+ *
+ * <p>
+ * The input is a plain text file with lines separated by newline characters.
+ *
+ * <p>
+ * Usage: <code>WordCount <text path> <result path></code><br>
+ * If no parameters are provided, the program is run with default data from {@link WordCountData}.
+ *
+ * <p>
+ * This example shows how to:
+ * <ul>
+ * <li>run a regular Storm program locally on Flink
+ * </ul>
+ */
+public class StormWordCountLocal {
+ public final static String topologyId = "Streaming WordCount";
+
+ // *************************************************************************
+ // PROGRAM
+ // *************************************************************************
+
+ public static void main(final String[] args) throws Exception {
+
+ if(!WordCountTopology.parseParameters(args)) {
+ return;
+ }
+
+ // build Topology the Storm way
+ final FlinkTopologyBuilder builder = WordCountTopology.buildTopology();
+
+ // execute program locally
+ final FlinkLocalCluster cluster = FlinkLocalCluster.getLocalCluster();
+ cluster.submitTopology(topologyId, null, builder.createTopology());
+
+ Utils.sleep(5 * 1000);
--- End diff --
Storm submitts topologies always in a non-blocking way. This behavior is mimiced in the compatibility layer.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-101174700
Thanks for the updates.
Although the examples are self-explanatory could you please add a readme on a minimal use case and explaining the features that you currently cover, please?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27808823
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkTopologyContext.java ---
@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import backtype.storm.generated.ComponentCommon;
+import backtype.storm.generated.StormTopology;
+import backtype.storm.hooks.ITaskHook;
+import backtype.storm.metric.api.CombinedMetric;
+import backtype.storm.metric.api.ICombiner;
+import backtype.storm.metric.api.IMetric;
+import backtype.storm.metric.api.IReducer;
+import backtype.storm.metric.api.ReducedMetric;
+import backtype.storm.state.ISubscribedState;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.tuple.Fields;
+import clojure.lang.Atom;
+
+
+
+
+
+/**
+ * {@link FlinkTopologyContext} is a {@link TopologyContext} that overwrites certain method that are not applicable when
+ * a Storm topology is executed within Flink.
+ *
+ * @author mjsax
+ */
+public class FlinkTopologyContext extends TopologyContext {
+ /**
+ * The topology belonging to this context.
+ */
+ private final StormTopology topology;
+
+
+
+ /**
+ * TODO remove unused parameters...
+ *
+ * @param topology
+ * @param stormConf
+ * @param taskToComponent
+ * @param componentToSortedTasks
+ * @param componentToStreamToFields
+ * @param stormId
+ * @param codeDir
+ * @param pidDir
+ * @param taskId
+ * @param workerPort
+ * @param workerTasks
+ * @param defaultResources
+ * @param userResources
+ * @param executorData
+ * @param registeredMetrics
+ * @param openOrPrepareWasCalled
+ */
+ public FlinkTopologyContext(final StormTopology topology, @SuppressWarnings("rawtypes") final Map stormConf,
+ final Map<Integer, String> taskToComponent, final Map<String, List<Integer>> componentToSortedTasks,
+ final Map<String, Map<String, Fields>> componentToStreamToFields, final String stormId, final String codeDir,
+ final String pidDir, final Integer taskId, final Integer workerPort, final List<Integer> workerTasks,
+ final Map<String, Object> defaultResources, final Map<String, Object> userResources,
+ final Map<String, Object> executorData, @SuppressWarnings("rawtypes") final Map registeredMetrics,
+ final Atom openOrPrepareWasCalled) {
+ super(topology, stormConf, taskToComponent, componentToSortedTasks, componentToStreamToFields, stormId,
+ codeDir, pidDir, taskId, workerPort, workerTasks, defaultResources, userResources, executorData,
+ registeredMetrics, openOrPrepareWasCalled);
+ this.topology = topology;
+ }
+
+
+
+ @Override
+ public ComponentCommon getComponentCommon(final String componentId) {
+ return super.getComponentCommon(componentId);
+ // final Bolt bolt = this.topology.get_bolts().get(componentId);
+ // if(bolt != null) {
+ // return bolt.get_common();
+ // }
+ // final SpoutSpec spoutSpec = this.topology.get_spouts().get(componentId);
--- End diff --
Couple of things seems a bit like overkill here. :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r30117434
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/test/java/org/apache/flink/stormcompatibility/wrappers/StormWrapperSetupHelperTest.java ---
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.stormcompatibility.wrappers;
+
+import static org.mockito.Mockito.mock;
+
+import java.util.Random;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import backtype.storm.topology.IComponent;
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.tuple.Fields;
+
+
+
+
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(StormWrapperSetupHelper.class)
+public class StormWrapperSetupHelperTest {
+ private Random r;
+
+
+
+ @Before
+ public void prepareTest() {
+ final long seed = System.currentTimeMillis();
+ this.r = new Random(seed);
+ System.out.println("Test seed: " + seed);
--- End diff --
I'd suggest "info"
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27822848
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/wrappers/StormSpoutWrapper.java ---
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.wrappers;
+
+import org.apache.flink.api.java.tuple.Tuple25;
+
+import backtype.storm.topology.IRichSpout;
+
+
+
+
+
--- End diff --
Why? Can it be a problem?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/573#issuecomment-105849950
Anything blocking this pull request, or can it be merged?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: Flink Storm compatibility
Posted by mjsax <gi...@git.apache.org>.
Github user mjsax commented on a diff in the pull request:
https://github.com/apache/flink/pull/573#discussion_r27822694
--- Diff: flink-staging/flink-streaming/flink-storm-compatibility/src/main/java/org/apache/flink/stormcompatibility/api/FlinkOutputFieldsDeclarer.java ---
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.stormcompatibility.api;
+
+import java.util.List;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+
+import backtype.storm.topology.IRichBolt;
+import backtype.storm.topology.IRichSpout;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.utils.Utils;
+
+
+
+
+
+/**
+ * {@link FlinkOutputFieldsDeclarer} is used to get the declared output schema of a {@link IRichSpout spout} or
+ * {@link IRichBolt bolt}.<br />
+ * <br />
+ * <strong>CAUTION: Currently, Flink does only support the default output stream. Furthermore, direct emit is not
+ * supported.</strong>
+ *
+ * @author mjsax
+ */
+final class FlinkOutputFieldsDeclarer implements OutputFieldsDeclarer {
+ private Fields outputSchema;
+
+ @Override
+ public void declare(final Fields fields) {
+ this.declareStream(Utils.DEFAULT_STREAM_ID, false, fields);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * Direct emit is no supported by Flink. Parameter {@code direct} must be {@code false}.
+ *
+ * @throws UnsupportedOperationException
+ * if {@code direct} is {@code true}
+ */
+ @Override
+ public void declare(final boolean direct, final Fields fields) {
+ this.declareStream(Utils.DEFAULT_STREAM_ID, direct, fields);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * Currently, Flink only supports the default output stream. Thus, pareamter {@code streamId} must be equals to
+ * {@link Utils#DEFAULT_STREAM_ID}.
+ *
+ * @throws UnsupportedOperationException
+ * if {@code streamId} is not equal to {@link Utils#DEFAULT_STREAM_ID}
+ */
+ @Override
+ public void declareStream(final String streamId, final Fields fields) {
+ this.declareStream(streamId, false, fields);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * Currently, Flink only supports the default output stream. Thus, pareamter {@code streamId} must be equals to
+ * {@link Utils#DEFAULT_STREAM_ID}. Furthermore, direct emit is no supported by Flink and parameter {@code direct}
+ * must be {@code false}.
+ *
+ * @throws UnsupportedOperationException
+ * if {@code streamId} is not equal to {@link Utils#DEFAULT_STREAM_ID} or {@code direct} is {@code true}
+ */
+ @Override
+ public void declareStream(final String streamId, final boolean direct, final Fields fields) {
+ if(!Utils.DEFAULT_STREAM_ID.equals(streamId)) {
+ throw new UnsupportedOperationException("Currently, only the default output stream is supported by Flink");
+ }
+ if(direct) {
+ throw new UnsupportedOperationException("Direct emit is not supported by Flink");
+ }
+
+ this.outputSchema = fields;
+ }
+
+ /**
+ * Returns {@link TypeInformation} for the declared output schema. If no or an empty output schema was declared,
+ * {@code null} is returned.
+ *
+ * @return output type information for the declared output schema; or {@code null} if no output schema was declared
+ *
+ * @throws IllegalArgumentException
+ * if more then 25 attributes are declared
+ */
+ public TypeInformation<?> getOutputType() throws IllegalArgumentException {
+ if((this.outputSchema == null) || (this.outputSchema.size() == 0)) {
+ return null;
+ }
+
+ Tuple t;
+ final int numberOfAttributes = this.outputSchema.size();
+
+ if(numberOfAttributes == 1) {
+ return TypeExtractor.getForClass(Object.class);
+ } else if(numberOfAttributes <= 25) {
+ try {
+ t = Tuple.getTupleClass(numberOfAttributes).newInstance();
+ } catch(final InstantiationException e) {
+ throw new RuntimeException(e);
+ } catch(final IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ throw new IllegalArgumentException("Flink supports only a maximum number of 25 attributes.");
+ }
+
+ // TODO: declare only key fields as DummyComparable
+ for(int i = 0; i < numberOfAttributes; ++i) {
+ t.setField(new DummyComparable(), i);
+ }
+
+ return TypeExtractor.getForObject(t);
+ }
+
+ /**
+ * {@link DummyComparable} is a {@link Comparable} helper class that is used to get the correct
+ * {@link TypeInformation} from {@link TypeExtractor} within {@link #getOutputType()}. If key fields are not
+ * comparable, Flink cannot use them and will throw an exception.
+ *
--- End diff --
As the comment says: "If key fields are NOT comparable, Flink cannot use them and will throw an exception."
Any suggestion for a better name?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---