fixed bug in flink wrapper
@@ -44,8 +44,7 @@ public abstract class FlinkRuntime<B extends BindingParams<I>, I extends Invocab
protected abstract void appendExecutionConfig(DataStream<Map<String, Object>>... convertedStream);
- public void run()
- {
+ public void run() {
try {
env.execute(bindingParams.getGraph().getElementId());
@@ -70,6 +69,7 @@ public abstract class FlinkRuntime<B extends BindingParams<I>, I extends Invocab
* This method takes the i's input stream and creates a source for the flink graph
* Currently just kafka is supported as a protocol
* TODO Add also jms support
+ *
* @param i
* @return
*/
@@ -84,18 +84,17 @@ public abstract class FlinkRuntime<B extends BindingParams<I>, I extends Invocab
(), getProperties(protocol));
} else {
return null;
- }} else {
+ }
+ } else {
return null;
}
}
@Override
public void prepareRuntime() throws SpRuntimeException {
- try {
if (debug) {
this.env = StreamExecutionEnvironment.createLocalEnvironment();
- }
- else {
+ } else {
this.env = StreamExecutionEnvironment
.createRemoteEnvironment(config.getHost(), config.getPort(), config.getJarFile());
}
@@ -122,6 +121,21 @@ public abstract class FlinkRuntime<B extends BindingParams<I>, I extends Invocab
appendExecutionConfig(messageStream1);
}
+ }
+
+ @Override
+ public void postDiscard() throws SpRuntimeException {
+
+ }
+
+ @Override
+ public void bindRuntime() throws SpRuntimeException {
+ try {
+ prepareRuntime();
+
+ Thread thread = new Thread(this);
+ thread.start();
+
// TODO find a better solution
// The loop waits until the job is deployed
// When the deployment takes longer then 60 seconds it returns false
@@ -152,19 +166,6 @@ public abstract class FlinkRuntime<B extends BindingParams<I>, I extends Invocab
}
}
- @Override
- public void postDiscard() throws SpRuntimeException {
-
- }
-
- @Override
- public void bindRuntime() throws SpRuntimeException {
- prepareRuntime();
-
- Thread thread = new Thread(this);
- thread.start();
- }
-
@Override
public void discardRuntime() throws SpRuntimeException {
FlinkJobController ctrl = new FlinkJobController(config.getHost(), config.getPort());
@@ -180,6 +181,7 @@ public abstract class FlinkRuntime<B extends BindingParams<I>, I extends Invocab
/**
* This method can be called in case additional environment settings should be applied to the runtime.
+ *
* @param env The Stream Execution environment
*/
public void appendEnvironmentConfig(StreamExecutionEnvironment env) {
