flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Boris Lublinsky <boris.lublin...@lightbend.com>
Subject Executing Flink server From IntelliJ
Date Fri, 14 Jul 2017 13:43:46 GMT
Hi, 
I am trying to upgrade my project from Flink 1.2 to 1.3 and getting problems while trying
to run Flink server from my Intellij project. The code

// Execute on the local Flink server - to test queariable state
def executeServer() : Unit = {

  // We use a mini cluster here for sake of simplicity, because I don't want
  // to require a Flink installation to run this demo. Everything should be
  // contained in this JAR.

  val port = 6124
  val parallelism = 4
  val config = new Configuration()
  config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, port)
  config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1)
  config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, parallelism)
  // In a non MiniCluster setup queryable state is enabled by default.
  config.setBoolean(QueryableStateOptions.SERVER_ENABLE, true)

  // Create a local Flink server
  val flinkCluster = new LocalFlinkMiniCluster(config, false)
  try {
    // Start server and create environment
    flinkCluster.start(true);
    val env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", port, parallelism)
    // Build Graph
    buildGraph(env)
    env.execute()
    val jobGraph = env.getStreamGraph.getJobGraph
    // Submit to the server and wait for completion
    flinkCluster.submitJobAndWait(jobGraph, false)
  } catch {
    case e: Exception => e.printStackTrace()
  }
}
Worked on version 1.2, but on 1.3 I am getting

08:41:29,179 INFO  org.apache.flink.runtime.minicluster.FlinkMiniCluster         - Starting
FlinkMiniCluster.
08:41:29,431 INFO  akka.event.slf4j.Slf4jLogger                                  - Slf4jLogger
started
08:41:29,498 INFO  Remoting                                                      - Starting
remoting
08:41:29,730 INFO  Remoting                                                      - Remoting
started; listening on addresses :[akka.tcp://flink@localhost:6124]
08:41:29,762 INFO  org.apache.flink.runtime.blob.BlobServer                      - Created
BLOB server storage directory /var/folders/3m/52z04fgs3hq88mzft9l0fsrm0000gn/T/blobStore-4e626961-9155-47e9-b1b8-f835a8435cfc
08:41:29,765 INFO  org.apache.flink.runtime.blob.BlobServer                      - Started
BLOB server at 0.0.0.0:54319 - max concurrent requests: 50 - max backlog: 1000
08:41:29,775 INFO  org.apache.flink.runtime.metrics.MetricRegistry               - No metrics
reporter configured, no metrics will be exposed/reported.
08:41:29,781 INFO  org.apache.flink.runtime.jobmanager.MemoryArchivist           - Started
memory archivist akka://flink/user/archive
08:41:29,786 INFO  org.apache.flink.runtime.jobmanager.JobManager                - Starting
JobManager at akka.tcp://flink@localhost:6124/user/jobmanager.
08:41:29,787 INFO  org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService
 - Proposing leadership to contender org.apache.flink.runtime.jobmanager.JobManager@59cd5ef5
@ akka.tcp://flink@localhost:6124/user/jobmanager
08:41:29,796 INFO  akka.event.slf4j.Slf4jLogger                                  - Slf4jLogger
started
08:41:29,804 INFO  Remoting                                                      - Starting
remoting
08:41:29,813 INFO  Remoting                                                      - Remoting
started; listening on addresses :[akka.tcp://flink@localhost:54320]
08:41:29,825 INFO  akka.event.slf4j.Slf4jLogger                                  - Slf4jLogger
started
08:41:29,830 INFO  Remoting                                                      - Starting
remoting
08:41:29,836 INFO  Remoting                                                      - Remoting
started; listening on addresses :[akka.tcp://flink@localhost:54321]
08:41:29,846 INFO  org.apache.flink.runtime.jobmanager.JobManager                - JobManager
akka.tcp://flink@localhost:6124/user/jobmanager was granted leadership with leader session
ID Some(61d3ed9b-1c24-4bbf-99ef-c2a891613473).
08:41:29,847 INFO  org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderService
 - Received confirmation of leadership for leader akka.tcp://flink@localhost:6124/user/jobmanager
, session=61d3ed9b-1c24-4bbf-99ef-c2a891613473
08:41:29,850 INFO  org.apache.flink.runtime.taskexecutor.TaskManagerConfiguration  - Messages
have a max timeout of 10000 ms
08:41:29,851 INFO  org.apache.flink.runtime.clusterframework.standalone.StandaloneResourceManager
 - Received leader address but not running in leader ActorSystem. Cancelling registration.
08:41:29,855 INFO  org.apache.flink.runtime.taskexecutor.TaskManagerServices     - Temporary
file directory '/var/folders/3m/52z04fgs3hq88mzft9l0fsrm0000gn/T': total 464 GB, usable 353
GB (76.08% usable)
08:41:30,493 INFO  org.apache.flink.runtime.io.network.buffer.NetworkBufferPool  - Allocated
363 MB for network buffer pool (number of memory segments: 11634, bytes per segment: 32768).
08:41:30,506 INFO  org.apache.flink.runtime.io.network.NetworkEnvironment        - Starting
the network environment and its components.
08:41:30,508 INFO  org.apache.flink.runtime.taskexecutor.TaskManagerServices     - Limiting
managed memory to 1145 MB, memory will be allocated lazily.
08:41:30,512 INFO  org.apache.flink.runtime.io.disk.iomanager.IOManager          - I/O manager
uses directory /var/folders/3m/52z04fgs3hq88mzft9l0fsrm0000gn/T/flink-io-9ec461ff-086d-45cd-b69c-e9890217d8fc
for spill files.
08:41:30,514 INFO  org.apache.flink.runtime.metrics.MetricRegistry               - No metrics
reporter configured, no metrics will be exposed/reported.
08:41:30,561 INFO  org.apache.flink.runtime.filecache.FileCache                  - User file
cache uses directory /var/folders/3m/52z04fgs3hq88mzft9l0fsrm0000gn/T/flink-dist-cache-4c4e9bcf-5a66-43e7-b2e9-244f310c3c4c
08:41:30,570 INFO  org.apache.flink.runtime.filecache.FileCache                  - User file
cache uses directory /var/folders/3m/52z04fgs3hq88mzft9l0fsrm0000gn/T/flink-dist-cache-687f7d57-33d7-4df3-915f-481008043fef
08:41:30,575 INFO  org.apache.flink.runtime.taskmanager.TaskManager              - Starting
TaskManager actor at akka://flink/user/taskmanager#-1401663761.
08:41:30,576 INFO  org.apache.flink.runtime.taskmanager.TaskManager              - TaskManager
data connection information: 15d5b91a66be806304e6fe15fde8c0fe @ localhost (dataPort=-1)
08:41:30,576 INFO  org.apache.flink.runtime.taskmanager.TaskManager              - TaskManager
has 4 task slot(s).
08:41:30,578 INFO  org.apache.flink.runtime.taskmanager.TaskManager              - Memory
usage stats: [HEAP: 391/838/3641 MB, NON HEAP: 25/26/-1 MB (used/committed/max)]
08:41:30,582 INFO  org.apache.flink.runtime.taskmanager.TaskManager              - Trying
to register at JobManager akka.tcp://flink@localhost:6124/user/jobmanager (attempt 1, timeout:
500 milliseconds)
08:41:30,729 INFO  org.apache.flink.runtime.jobmanager.JobManager                - Task Manager
Registration but not connected to ResourceManager
08:41:30,732 INFO  org.apache.flink.runtime.instance.InstanceManager             - Registered
TaskManager at localhost (akka.tcp://flink@localhost:54321/user/taskmanager) as 38047c3fc643910d58ecc414e8233f78.
Current number of registered hosts is 1. Current number of alive task slots is 4.
08:41:30,741 INFO  org.apache.flink.runtime.taskmanager.TaskManager              - Successful
registration at JobManager (akka.tcp://flink@localhost:6124/user/jobmanager), starting network
stack and library cache.
08:41:30,743 INFO  org.apache.flink.runtime.taskmanager.TaskManager              - Determined
BLOB server address to be localhost/127.0.0.1:54319. Starting BLOB cache.
08:41:30,745 INFO  org.apache.flink.runtime.blob.BlobCache                       - Created
BLOB cache storage directory /var/folders/3m/52z04fgs3hq88mzft9l0fsrm0000gn/T/blobStore-122d4c13-34d1-4c01-8a50-f6dfdae8b06b
08:41:30,996 INFO  org.apache.flink.streaming.api.environment.RemoteStreamEnvironment  - Running
remotely at localhost:6124
08:41:31,085 INFO  org.apache.flink.client.program.StandaloneClusterClient       - Starting
client actor system.
08:41:31,087 INFO  org.apache.flink.runtime.util.LeaderRetrievalUtils            - Trying
to select the network interface and address to use by connecting to the leading JobManager.
08:41:31,087 INFO  org.apache.flink.runtime.util.LeaderRetrievalUtils            - TaskManager
will try to connect for 10000 milliseconds before falling back to heuristics
08:41:31,088 INFO  org.apache.flink.runtime.net.ConnectionUtils                  - Retrieved
new target address localhost/127.0.0.1:6124.
08:41:31,100 INFO  akka.event.slf4j.Slf4jLogger                                  - Slf4jLogger
started
08:41:31,103 INFO  Remoting                                                      - Starting
remoting
08:41:31,108 INFO  Remoting                                                      - Remoting
started; listening on addresses :[akka.tcp://flink@localhost:54324]
08:41:31,108 INFO  org.apache.flink.client.program.StandaloneClusterClient       - Submitting
job with JobID: 74abb7674b9522ad3a204a1315cf609e. Waiting for job completion.
Submitting job with JobID: 74abb7674b9522ad3a204a1315cf609e. Waiting for job completion.
08:41:31,113 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Disconnect
from JobManager null.
08:41:31,116 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Received
SubmitJobAndWait(JobGraph(jobId: 74abb7674b9522ad3a204a1315cf609e)) but there is no connection
to a JobManager yet.
08:41:31,116 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Received
job Flink Streaming Job (74abb7674b9522ad3a204a1315cf609e).
08:41:31,125 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Connect
to JobManager Actor[akka.tcp://flink@localhost:6124/user/jobmanager#-297192771].
08:41:31,126 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Connected
to JobManager at Actor[akka.tcp://flink@localhost:6124/user/jobmanager#-297192771] with leader
session id 00000000-0000-0000-0000-000000000000.
Connected to JobManager at Actor[akka.tcp://flink@localhost:6124/user/jobmanager#-297192771]
with leader session id 00000000-0000-0000-0000-000000000000.
08:41:31,126 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Sending
message to JobManager akka.tcp://flink@localhost:6124/user/jobmanager to submit job Flink
Streaming Job (74abb7674b9522ad3a204a1315cf609e) and wait for progress
08:41:31,128 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Upload
jar files to job manager akka.tcp://flink@localhost:6124/user/jobmanager.
08:41:31,129 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Submit
job to the job manager akka.tcp://flink@localhost:6124/user/jobmanager.
08:41:31,146 WARN  org.apache.flink.runtime.jobmanager.JobManager                - Discard
message LeaderSessionMessage(00000000-0000-0000-0000-000000000000,SubmitJob(JobGraph(jobId:
74abb7674b9522ad3a204a1315cf609e),EXECUTION_RESULT_AND_STATE_CHANGES)) because the expected
leader session ID 61d3ed9b-1c24-4bbf-99ef-c2a891613473 did not equal the received leader session
ID 00000000-0000-0000-0000-000000000000.
08:42:30,381 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Terminate
JobClientActor.
08:42:30,382 INFO  org.apache.flink.runtime.client.JobSubmissionClientActor      - Disconnect
from JobManager Actor[akka.tcp://flink@localhost:6124/user/jobmanager#-297192771].
08:42:30,391 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Shutting
down remote daemon.
08:42:30,392 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Remote
daemon shut down; proceeding with flushing remote transports.
08:42:30,411 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Remoting
shut down.
org.apache.flink.client.program.ProgramInvocationException: The program execution failed:
Couldn't retrieve the JobExecutionResult from the JobManager.
	at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:478)
	at org.apache.flink.client.program.StandaloneClusterClient.submitJob(StandaloneClusterClient.java:105)
	at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:442)
	at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:434)
	at org.apache.flink.streaming.api.environment.RemoteStreamEnvironment.executeRemotely(RemoteStreamEnvironment.java:212)
	at org.apache.flink.streaming.api.environment.RemoteStreamEnvironment.execute(RemoteStreamEnvironment.java:176)
	at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1499)
	at org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.scala:629)
	at com.lightbend.modelServer.ModelServingKeyedJob$.executeServer(ModelServingKeyedJob.scala:66)
	at com.lightbend.modelServer.ModelServingKeyedJob$.main(ModelServingKeyedJob.scala:39)
	at com.lightbend.modelServer.ModelServingKeyedJob.main(ModelServingKeyedJob.scala)
Caused by: org.apache.flink.runtime.client.JobExecutionException: Couldn't retrieve the JobExecutionResult
from the JobManager.
	at org.apache.flink.runtime.client.JobClient.awaitJobResult(JobClient.java:309)
	at org.apache.flink.runtime.client.JobClient.submitJobAndWait(JobClient.java:396)
	at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:467)
	... 10 more
Caused by: org.apache.flink.runtime.client.JobClientActorSubmissionTimeoutException: Job submission
to the JobManager timed out. You may increase 'akka.client.timeout' in case the JobManager
needs more time to configure and confirm the job submission.
	at org.apache.flink.runtime.client.JobSubmissionClientActor.handleCustomMessage(JobSubmissionClientActor.java:119)
	at org.apache.flink.runtime.client.JobClientActor.handleMessage(JobClientActor.java:251)
	at org.apache.flink.runtime.akka.FlinkUntypedActor.handleLeaderSessionID(FlinkUntypedActor.java:89)
	at org.apache.flink.runtime.akka.FlinkUntypedActor.onReceive(FlinkUntypedActor.java:68)
	at akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:167)
	at akka.actor.Actor$class.aroundReceive(Actor.scala:467)
	at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:97)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
	at akka.actor.ActorCell.invoke(ActorCell.scala:487)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)
	at akka.dispatch.Mailbox.run(Mailbox.scala:220)
	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397)
	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
08:42:30,424 INFO  org.apache.flink.runtime.blob.BlobCache                       - Shutting
down BlobCache
08:42:30,433 INFO  org.apache.flink.runtime.blob.BlobServer                      - Stopped
BLOB server at 0.0.0.0:54319
08:42:30,434 INFO  org.apache.flink.runtime.io.disk.iomanager.IOManager          - I/O manager
removed spill file directory /var/folders/3m/52z04fgs3hq88mzft9l0fsrm0000gn/T/flink-io-9ec461ff-086d-45cd-b69c-e9890217d8fc

Process finished with exit code 0

Any help will be appreciated



Boris Lublinsky
FDP Architect
boris.lublinsky@lightbend.com
https://www.lightbend.com/


Mime
View raw message