Skip to content

Commit

Permalink
[hotfix] Add DefaultActorSystemLoader which returns the given ActorSy…
Browse files Browse the repository at this point in the history
…stem
  • Loading branch information
tillrohrmann committed Jun 22, 2018
1 parent b946c5e commit ccfe9bb
Show file tree
Hide file tree
Showing 4 changed files with 85 additions and 14 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
/*
* 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:https://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.client.program;

import org.apache.flink.util.FlinkException;

import akka.actor.ActorSystem;

/**
* Interface which allows to load an {@link ActorSystem}.
*/
public interface ActorSystemLoader extends AutoCloseable {

/**
* Get an {@link ActorSystem}.
*
* @return {@link ActorSystem}
* @throws FlinkException
*/
ActorSystem get() throws FlinkException;
}
Original file line number Diff line number Diff line change
Expand Up @@ -212,20 +212,6 @@ public ClusterClient(
// Startup & Shutdown
// ------------------------------------------------------------------------

/**
* Interface which allows to load an {@link ActorSystem}.
*/
public interface ActorSystemLoader extends AutoCloseable {

/**
* Get an {@link ActorSystem}.
*
* @return {@link ActorSystem}
* @throws FlinkException
*/
ActorSystem get() throws FlinkException;
}

/**
* Utility class to lazily instantiate an {@link ActorSystem}.
*/
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* 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:https://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.client.program;

import akka.actor.ActorSystem;

import javax.annotation.Nonnull;

/**
* {@link ActorSystemLoader} which simply returns the given {@link ActorSystem} at
* construction time.
*/
public class DefaultActorSystemLoader implements ActorSystemLoader {

@Nonnull
private final ActorSystem actorSystem;

public DefaultActorSystemLoader(@Nonnull ActorSystem actorSystem) {
this.actorSystem = actorSystem;
}

@Override
public ActorSystem get() {
return actorSystem;
}

@Override
public void close() throws Exception {
// noop
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobSubmissionResult;
import org.apache.flink.client.program.ActorSystemLoader;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.configuration.Configuration;
Expand Down

0 comments on commit ccfe9bb

Please sign in to comment.