title: Flink RabbitMQ Correlation ID
body: <p>I´m trying to understand how to use high availability in Flink with RabbiMQ, using Correlation Ids and enabling Checkpoints, but it is no working. My Producer code:</p>
  
  <pre><code>connection = factory.newConnection();
  Channel channel = connection.createChannel();
  
  String corrId = java.util.UUID.randomUUID().toString();
  BasicProperties props = new AMQP.BasicProperties().builder().correlationId(corrId).build();
  
  channel.queueDeclare("flink-poc", true, false, false, null);
  MessageQueue queue = new MessageQueue(500); //Queue of messages to be sent to rabbitmq
  Message msg = null;
  while ((msg = queue.takeMessage()) != null) 
      channel.basicPublish("", "flink-poc", props, mapper.writeValueAsBytes(msg));
  channel.close();
  connection.close();
  </code></pre>
  
  <p>Consumer code:</p>
  
  <pre><code>StreamExecutionEnvironment env = StreamExecutionEnvironment
              .getExecutionEnvironment();
  env.setBufferTimeout(100);
  env.enableCheckpointing(1000,CheckpointingMode.EXACTLY_ONCE); // start a checkpoint every 1000 ms
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  
  RabbitSource&lt;Message&gt; rabbitSource = new RabbitSource&lt;Message&gt;(Host, 5672, username, pass,  "flink-poc", VirtualHost, true, schema);
  messages = env.addSource(rabbitSource, TypeInformation.of(Message.class)
  </code></pre>
  
  <p>With this code, no ACK is returned to RabbitMQ. Any help is appreciated.</p>
  
  <p><strong>Update:</strong> With a prefetch (<code>channel.basicQos(15);</code>), it starts working, but with a very low rate. Is there any way to improve it? If <code>setStreamTimeCharacteristic</code> is deactivated, the results are ordered and improves the rate by 10 times, but still very low rate, how is this possible?</p>