(DEPRECATED) Apache Flink User Mailing List archive.

回复：Re: Using RabbitMQ Sinks

Classic

List

Threaded

2 messages Options

rimin515

回复：Re: Using RabbitMQ Sinks

Hi,

i use the code follow,

def main(args:Array[String]):Unit={

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.enableCheckpointing(1000)

val host = ""

val quene = ""

val port =***

val user = ""

val pass = ""

//rmqConnectionConfig

val rmqConnectionConfig= new RMQConnectionConfig.Builder().setHost(host).setPort(port).setUserName(user).setPassword(pass).setVirtualHost("/").build()

///

//println(rmqConnectionConfig.getHost)

val streamrdd=env.addSource(new RMQSource(rmqConnectionConfig,quene,false,new SimpleStringSchema))

streamrdd.print()

println("**************************************")

env.execute("read rabbitmq")

}

it can work,and read the data. i think you maybe have a wrong rabbitmq config,or use the wrong package,

for example,i use the 1.1-SNAPSHOT.

-------------------------------

----- 原始邮件 -----
发件人：Paul Joireman <[hidden email]>
收件人："[hidden email]" <[hidden email]>
主题：Re: Using RabbitMQ Sinks
日期：2016年08月08日 23点27分

Robert,

It looks like the root cause exception is:

com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - parameters for queue 'flink-sink' in vhost '/' not equivalent, class-id=50, method-id=10)

The full execution trace is shown below and occurs when I try to execute my process. In the process, I initially create a connection to RMQ using the RQQConnectionConfig.Builder() and this works fine if I read using a SimpleStringSchema but re-using the same connection configuration for the sink as follows:

msgs.addSink(new RMQSink<String>(rmqConnConfig, ALERT_SINK_QUEUE, new SimpleStringSchema()));

Where msgs are Strings.

Paul

Connected to the target VM, address: '127.0.0.1:42208', transport: 'socket'

10:13:50,555 INFO org.apache.flink.api.java.typeutils.TypeExtractor - class com.physiq.alert.AlertMessageIn is not a valid POJO type

10:13:57,035 INFO org.apache.flink.streaming.api.environment.LocalStreamEnvironment - Running job on local embedded Flink mini cluster

10:13:58,100 INFO org.apache.flink.runtime.minicluster.FlinkMiniCluster - Starting FlinkMiniCluster.

10:13:58,580 INFO akka.event.slf4j.Slf4jLogger - Slf4jLogger started

10:13:58,614 INFO org.apache.flink.runtime.blob.BlobServer - Created BLOB server storage directory /tmp/blobStore-b19e22df-4636-4e96-bb45-db70c0bcc7e1

10:13:58,620 INFO org.apache.flink.runtime.blob.BlobServer - Started BLOB server at 0.0.0.0:34360 - max concurrent requests: 50 - max backlog: 1000

10:13:58,635 INFO org.apache.flink.runtime.checkpoint.savepoint.SavepointStoreFactory - Using job manager savepoint state backend.

10:13:58,642 INFO org.apache.flink.runtime.metrics.MetricRegistry - No metrics reporter configured, no metrics will be exposed/reported.

10:13:58,655 INFO org.apache.flink.runtime.jobmanager.MemoryArchivist - Started memory archivist akka://flink/user/archive_1

10:13:58,657 INFO org.apache.flink.runtime.jobmanager.JobManager - Starting JobManager at akka://flink/user/jobmanager_1.

10:13:58,678 INFO org.apache.flink.runtime.clusterframework.standalone.StandaloneResourceManager - Trying to associate with JobManager leader akka://flink/user/jobmanager_1

10:13:58,683 INFO org.apache.flink.runtime.taskmanager.TaskManager - Messages between TaskManager and JobManager have a max timeout of 10000 milliseconds

10:13:58,703 INFO org.apache.flink.runtime.taskmanager.TaskManager - Temporary file directory '/tmp': total 102 GB, usable 9 GB (8.82% usable)

10:13:58,844 INFO org.apache.flink.runtime.io.network.buffer.NetworkBufferPool - Allocated 64 MB for network buffer pool (number of memory segments: 2048, bytes per segment: 32768).

10:13:58,846 INFO org.apache.flink.runtime.taskmanager.TaskManager - Limiting managed memory to 549 MB, memory will be allocated lazily.

10:14:00,163 INFO org.apache.flink.runtime.io.disk.iomanager.IOManager - I/O manager uses directory /tmp/flink-io-e6686803-fb13-4e2f-a4df-cb6ca465eb2b for spill files.

10:14:00,167 INFO org.apache.flink.runtime.jobmanager.JobManager - JobManager akka://flink/user/jobmanager_1 was granted leadership with leader session ID None.

10:14:00,239 INFO org.apache.flink.runtime.filecache.FileCache - User file cache uses directory /tmp/flink-dist-cache-2e8256c3-532c-45b6-ad2d-fa7b43846a6e

10:14:00,239 INFO org.apache.flink.runtime.clusterframework.standalone.StandaloneResourceManager - Resource Manager associating with leading JobManager Actor[akka://flink/user/jobmanager_1#-1689614413] - leader session null

10:14:00,495 INFO org.apache.flink.runtime.taskmanager.TaskManager - Starting TaskManager actor at akka://flink/user/taskmanager_1#2008243751.

10:14:00,496 INFO org.apache.flink.runtime.taskmanager.TaskManager - TaskManager data connection information: localhost (dataPort=58335)

10:14:00,496 INFO org.apache.flink.runtime.taskmanager.TaskManager - TaskManager has 4 task slot(s).

10:14:00,498 INFO org.apache.flink.runtime.taskmanager.TaskManager - Memory usage stats: [HEAP: 77/195/1701 MB, NON HEAP: 22/33/214 MB (used/committed/max)]

10:14:00,506 INFO org.apache.flink.runtime.taskmanager.TaskManager - Trying to register at JobManager akka://flink/user/jobmanager_1 (attempt 1, timeout: 500 milliseconds)

10:14:00,508 INFO org.apache.flink.runtime.clusterframework.standalone.StandaloneResourceManager - TaskManager ResourceID{resourceId='8ca61ba2f8741d54ce649bd0d525d50c'} has started.

10:14:00,511 INFO org.apache.flink.runtime.instance.InstanceManager - Registered TaskManager at localhost (akka://flink/user/taskmanager_1) as db51cf9997f8059543464810ffaffea3. Current number of registered hosts is 1. Current number of alive task slots is 4.

10:14:00,517 INFO org.apache.flink.runtime.taskmanager.TaskManager - Successful registration at JobManager (akka://flink/user/jobmanager_1), starting network stack and library cache.

10:14:00,534 INFO org.apache.flink.runtime.taskmanager.TaskManager - Determined BLOB server address to be localhost/127.0.0.1:34360. Starting BLOB cache.

10:14:00,535 INFO org.apache.flink.runtime.blob.BlobCache - Created BLOB cache storage directory /tmp/blobStore-1a9647ad-11f0-40de-b24c-0e2c57c030d0

10:14:00,535 INFO org.apache.flink.runtime.metrics.MetricRegistry - No metrics reporter configured, no metrics will be exposed/reported.

10:14:00,549 INFO org.apache.flink.runtime.client.JobClientActor - Received job Flink Streaming Job (6d641f666b26088a843355ff3b0b1705).

10:14:00,549 INFO org.apache.flink.runtime.client.JobClientActor - Could not submit job Flink Streaming Job (6d641f666b26088a843355ff3b0b1705), because there is no connection to a JobManager.

10:14:00,551 INFO org.apache.flink.runtime.client.JobClientActor - Disconnect from JobManager null.

10:14:00,557 INFO org.apache.flink.runtime.client.JobClientActor - Connect to JobManager Actor[akka://flink/user/jobmanager_1#-1689614413].

10:14:00,557 INFO org.apache.flink.runtime.client.JobClientActor - Connected to new JobManager akka://flink/user/jobmanager_1.

10:14:00,557 INFO org.apache.flink.runtime.client.JobClientActor - Sending message to JobManager akka://flink/user/jobmanager_1 to submit job Flink Streaming Job (6d641f666b26088a843355ff3b0b1705) and wait for progress

10:14:00,558 INFO org.apache.flink.runtime.client.JobClientActor - Upload jar files to job manager akka://flink/user/jobmanager_1.

10:14:00,560 INFO org.apache.flink.runtime.client.JobClientActor - Submit job to the job manager akka://flink/user/jobmanager_1.

10:14:00,563 INFO org.apache.flink.runtime.jobmanager.JobManager - Submitting job 6d641f666b26088a843355ff3b0b1705 (Flink Streaming Job).

10:14:00,568 INFO org.apache.flink.runtime.jobmanager.JobManager - Using restart strategy NoRestartStrategy for 6d641f666b26088a843355ff3b0b1705.

10:14:00,628 INFO org.apache.flink.runtime.jobmanager.JobManager - Scheduling job 6d641f666b26088a843355ff3b0b1705 (Flink Streaming Job).

10:14:00,628 INFO org.apache.flink.runtime.client.JobClientActor - Job was successfully submitted to the JobManager akka://flink/user/jobmanager_1.

10:14:00,631 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: Custom Source (1/1) (9ffba0e9f84adca163121d50f88e519e) switched from CREATED to SCHEDULED

10:14:00,631 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Job execution switched to status RUNNING.

08/08/2016 10:14:00 Job execution switched to status RUNNING.

10:14:00,632 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Source: Custom Source(1/1) switched to SCHEDULED

08/08/2016 10:14:00 Source: Custom Source(1/1) switched to SCHEDULED

10:14:00,633 INFO org.apache.flink.runtime.jobmanager.JobManager - Status of job 6d641f666b26088a843355ff3b0b1705 (Flink Streaming Job) changed to RUNNING.

10:14:00,637 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: Custom Source (1/1) (9ffba0e9f84adca163121d50f88e519e) switched from SCHEDULED to DEPLOYING

10:14:00,638 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Source: Custom Source(1/1) switched to DEPLOYING

08/08/2016 10:14:00 Source: Custom Source(1/1) switched to DEPLOYING

10:14:00,638 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Source: Custom Source (1/1) (attempt #0) to localhost

10:14:00,642 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (1/4) (4a13efc94d64d61532106fbd9bdfaedb) switched from CREATED to SCHEDULED

10:14:00,642 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(1/4) switched to SCHEDULED

08/08/2016 10:14:00 Map(1/4) switched to SCHEDULED

10:14:00,643 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (1/4) (4a13efc94d64d61532106fbd9bdfaedb) switched from SCHEDULED to DEPLOYING

10:14:00,643 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Map (1/4) (attempt #0) to localhost

10:14:00,643 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(1/4) switched to DEPLOYING

08/08/2016 10:14:00 Map(1/4) switched to DEPLOYING

10:14:00,648 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (2/4) (99136f14b2e32b44318b49b2ad39dde5) switched from CREATED to SCHEDULED

10:14:00,648 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (2/4) (99136f14b2e32b44318b49b2ad39dde5) switched from SCHEDULED to DEPLOYING

10:14:00,649 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(2/4) switched to SCHEDULED

08/08/2016 10:14:00 Map(2/4) switched to SCHEDULED

10:14:00,649 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(2/4) switched to DEPLOYING

08/08/2016 10:14:00 Map(2/4) switched to DEPLOYING

10:14:00,649 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Map (2/4) (attempt #0) to localhost

10:14:00,650 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (3/4) (029bc351981b5056208839ce988f0f3f) switched from CREATED to SCHEDULED

10:14:00,650 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(3/4) switched to SCHEDULED

10:14:00,650 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (3/4) (029bc351981b5056208839ce988f0f3f) switched from SCHEDULED to DEPLOYING

08/08/2016 10:14:00 Map(3/4) switched to SCHEDULED

10:14:00,655 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Map (3/4) (attempt #0) to localhost

10:14:00,656 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (4/4) (ae5e9324d04ffac7843317749a2e86dd) switched from CREATED to SCHEDULED

10:14:00,657 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (4/4) (ae5e9324d04ffac7843317749a2e86dd) switched from SCHEDULED to DEPLOYING

10:14:00,657 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(3/4) switched to DEPLOYING

08/08/2016 10:14:00 Map(3/4) switched to DEPLOYING

10:14:00,659 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Map (4/4) (attempt #0) to localhost

10:14:00,659 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(4/4) switched to SCHEDULED

08/08/2016 10:14:00 Map(4/4) switched to SCHEDULED

10:14:00,660 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(4/4) switched to DEPLOYING

08/08/2016 10:14:00 Map(4/4) switched to DEPLOYING

10:14:00,660 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (1/4) (b3a3e69c7dd082fc744b6ec791697e35) switched from CREATED to SCHEDULED

10:14:00,660 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(1/4) switched to SCHEDULED

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(1/4) switched to SCHEDULED

10:14:00,660 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (1/4) (b3a3e69c7dd082fc744b6ec791697e35) switched from SCHEDULED to DEPLOYING

10:14:00,661 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(1/4) switched to DEPLOYING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(1/4) switched to DEPLOYING

10:14:00,661 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Filter -> Map -> Sink: Unnamed (1/4) (attempt #0) to localhost

10:14:00,662 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (2/4) (54a627d8091077e6742cd74d754016a7) switched from CREATED to SCHEDULED

10:14:00,662 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(2/4) switched to SCHEDULED

10:14:00,662 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (2/4) (54a627d8091077e6742cd74d754016a7) switched from SCHEDULED to DEPLOYING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(2/4) switched to SCHEDULED

10:14:00,662 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(2/4) switched to DEPLOYING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(2/4) switched to DEPLOYING

10:14:00,662 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Filter -> Map -> Sink: Unnamed (2/4) (attempt #0) to localhost

10:14:00,663 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (3/4) (dafdfde1351e3eb9593fa227e8255b57) switched from CREATED to SCHEDULED

10:14:00,664 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(3/4) switched to SCHEDULED

10:14:00,664 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (3/4) (dafdfde1351e3eb9593fa227e8255b57) switched from SCHEDULED to DEPLOYING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(3/4) switched to SCHEDULED

10:14:00,664 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Filter -> Map -> Sink: Unnamed (3/4) (attempt #0) to localhost

10:14:00,664 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(3/4) switched to DEPLOYING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(3/4) switched to DEPLOYING

10:14:00,665 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (4/4) (7c219dda43bc53571d306fc0df4468fa) switched from CREATED to SCHEDULED

10:14:00,666 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(4/4) switched to SCHEDULED

10:14:00,666 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (4/4) (7c219dda43bc53571d306fc0df4468fa) switched from SCHEDULED to DEPLOYING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(4/4) switched to SCHEDULED

10:14:00,666 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Deploying Filter -> Map -> Sink: Unnamed (4/4) (attempt #0) to localhost

10:14:00,666 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(4/4) switched to DEPLOYING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(4/4) switched to DEPLOYING

10:14:00,674 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Source: Custom Source (1/1)

10:14:00,674 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Source: Custom Source (1/1)

10:14:00,678 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Map (1/4)

10:14:00,678 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Map (1/4)

10:14:00,683 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Map (2/4)

10:14:00,684 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Map (2/4)

10:14:00,684 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Map (3/4)

10:14:00,691 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Source: Custom Source (1/1) [DEPLOYING]

10:14:00,692 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Map (2/4) [DEPLOYING]

10:14:00,691 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Map (3/4)

10:14:00,691 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Map (4/4)

10:14:00,691 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Map (1/4) [DEPLOYING]

10:14:00,695 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Filter -> Map -> Sink: Unnamed (1/4)

10:14:00,695 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Map (3/4) [DEPLOYING]

10:14:00,695 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Filter -> Map -> Sink: Unnamed (2/4)

10:14:00,695 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Map (4/4)

10:14:00,696 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Filter -> Map -> Sink: Unnamed (1/4)

10:14:00,703 INFO org.apache.flink.runtime.taskmanager.Task - Map (2/4) switched to RUNNING

10:14:00,704 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Filter -> Map -> Sink: Unnamed (1/4) [DEPLOYING]

10:14:00,701 INFO org.apache.flink.runtime.taskmanager.Task - Source: Custom Source (1/1) switched to RUNNING

10:14:00,698 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Filter -> Map -> Sink: Unnamed (2/4)

10:14:00,709 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Filter -> Map -> Sink: Unnamed (2/4) [DEPLOYING]

10:14:00,699 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Filter -> Map -> Sink: Unnamed (3/4)

10:14:00,710 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Filter -> Map -> Sink: Unnamed (3/4)

10:14:00,698 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Map (4/4) [DEPLOYING]

10:14:00,711 INFO org.apache.flink.runtime.taskmanager.TaskManager - Received task Filter -> Map -> Sink: Unnamed (4/4)

10:14:00,710 INFO org.apache.flink.runtime.taskmanager.Task - Map (3/4) switched to RUNNING

10:14:00,715 INFO org.apache.flink.runtime.taskmanager.Task - Map (1/4) switched to RUNNING

10:14:00,719 INFO org.apache.flink.runtime.taskmanager.Task - Loading JAR files for task Filter -> Map -> Sink: Unnamed (4/4)

10:14:00,724 INFO org.apache.flink.runtime.taskmanager.Task - Filter -> Map -> Sink: Unnamed (1/4) switched to RUNNING

10:14:00,723 INFO org.apache.flink.runtime.taskmanager.Task - Filter -> Map -> Sink: Unnamed (2/4) switched to RUNNING

10:14:00,720 INFO org.apache.flink.runtime.taskmanager.Task - Map (4/4) switched to RUNNING

10:14:00,718 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (2/4) (99136f14b2e32b44318b49b2ad39dde5) switched from DEPLOYING to RUNNING

10:14:00,725 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: Custom Source (1/1) (9ffba0e9f84adca163121d50f88e519e) switched from DEPLOYING to RUNNING

10:14:00,725 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(2/4) switched to RUNNING

08/08/2016 10:14:00 Map(2/4) switched to RUNNING

10:14:00,726 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Source: Custom Source(1/1) switched to RUNNING

08/08/2016 10:14:00 Source: Custom Source(1/1) switched to RUNNING

10:14:00,730 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (3/4) (029bc351981b5056208839ce988f0f3f) switched from DEPLOYING to RUNNING

10:14:00,730 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(3/4) switched to RUNNING

08/08/2016 10:14:00 Map(3/4) switched to RUNNING

10:14:00,732 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (1/4) (4a13efc94d64d61532106fbd9bdfaedb) switched from DEPLOYING to RUNNING

10:14:00,732 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(1/4) switched to RUNNING

08/08/2016 10:14:00 Map(1/4) switched to RUNNING

10:14:00,733 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (1/4) (b3a3e69c7dd082fc744b6ec791697e35) switched from DEPLOYING to RUNNING

10:14:00,724 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Filter -> Map -> Sink: Unnamed (3/4) [DEPLOYING]

10:14:00,736 INFO org.apache.flink.runtime.taskmanager.Task - Registering task at network: Filter -> Map -> Sink: Unnamed (4/4) [DEPLOYING]

10:14:00,736 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(1/4) switched to RUNNING

10:14:00,750 INFO org.apache.flink.runtime.taskmanager.Task - Filter -> Map -> Sink: Unnamed (3/4) switched to RUNNING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(1/4) switched to RUNNING

10:14:00,750 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (2/4) (54a627d8091077e6742cd74d754016a7) switched from DEPLOYING to RUNNING

10:14:00,751 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Map (4/4) (ae5e9324d04ffac7843317749a2e86dd) switched from DEPLOYING to RUNNING

10:14:00,751 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(2/4) switched to RUNNING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(2/4) switched to RUNNING

10:14:00,752 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Map(4/4) switched to RUNNING

08/08/2016 10:14:00 Map(4/4) switched to RUNNING

10:14:00,751 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (3/4) (dafdfde1351e3eb9593fa227e8255b57) switched from DEPLOYING to RUNNING

10:14:00,757 INFO org.apache.flink.runtime.taskmanager.Task - Filter -> Map -> Sink: Unnamed (4/4) switched to RUNNING

10:14:00,770 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Filter -> Map -> Sink: Unnamed (4/4) (7c219dda43bc53571d306fc0df4468fa) switched from DEPLOYING to RUNNING

10:14:00,756 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,766 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,765 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(3/4) switched to RUNNING

10:14:00,771 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(3/4) switched to RUNNING

10:14:00,771 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,776 INFO org.apache.flink.runtime.client.JobClientActor - 08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(4/4) switched to RUNNING

08/08/2016 10:14:00 Filter -> Map -> Sink: Unnamed(4/4) switched to RUNNING

10:14:00,782 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,782 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,787 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,787 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,789 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,790 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,791 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,792 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,792 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,793 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,792 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,793 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,794 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,794 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,796 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,796 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,796 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,796 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,804 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,804 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,806 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,806 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,807 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,807 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,807 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,807 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,808 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,808 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:00,807 WARN org.apache.flink.streaming.runtime.tasks.StreamTask - No state backend has been specified, using default state backend (Memory / JobManager)

10:14:00,809 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - State backend is set to heap memory (checkpoint to jobmanager)

10:14:01,032 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - Timer service is shutting down.

10:14:01,034 INFO org.apache.flink.streaming.runtime.tasks.StreamTask - Timer service is shutting down.

10:14:01,037 ERROR org.apache.flink.streaming.api.operators.AbstractStreamOperator - Exception while closing user function while failing or canceling task

com.rabbitmq.client.AlreadyClosedException: channel is already closed due to channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - parameters for queue 'flink-sink' in vhost '/' not equivalent, class-id=50, method-id=10)

at com.rabbitmq.client.impl.AMQChannel.processShutdownSignal(AMQChannel.java:265)

at com.rabbitmq.client.impl.ChannelN.startProcessShutdownSignal(ChannelN.java:261)

at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:565)

at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:505)

at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:498)

at org.apache.flink.streaming.connectors.rabbitmq.RMQSink.close(RMQSink.java:114)

at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:45)

at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:107)

at org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:426)

at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:332)

at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)

at java.lang.Thread.run(Thread.java:745)