Hi Flink devs. We are trying to run a basic Flink job that subscribes to a RabbitMQ stream. Please note, this is the relatively recent feature of the

stream

data structure, as opposed to

queue

for RabbitMQ. We are setting up a basic job and have managed to get the connector working with the

queue

type, but not

stream

.

version: '3.3'
services:
  jobmanager:
    image: flink:1.17.0
    expose:
      - '8081'
    ports:
      - '8081:8081'
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
      - JOB_MANAGER_RPC_PORT=6123
      - MQ_IP=${MQ_IP}
      - MQ_PORT=${MQ_PORT}
      - MQ_Q=${MQ_Q}
      - MQ_UN=${MQ_UN}
      - MQ_PW=${MQ_PW}

  taskmanager:
    image: flink:1.17.0
    depends_on:
      - jobmanager
    expose:
      - '6121'
    ports:
      - '6121:6121'
    command: taskmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
      - JOB_MANAGER_RPC_PORT=6123
      - TASK_MANAGER_NUMBER_OF_TASK_SLOTS=4

networks:
  dev-network:
    driver: bridge

Connector version:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-rabbitmq</artifactId>
  <version>3.0.1-1.17</version>
</dependency>

Connecting to RabbitMQ:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
System.out.println("Connecting....................");

String MQ_IP = System.getenv("MQ_IP");
int MQ_PORT = Integer.parseInt(System.getenv("MQ_PORT"));
String MQ_UN = System.getenv("MQ_UN");
String MQ_PW = System.getenv("MQ_PW");
String MQ_Q = System.getenv("MQ_Q");

// RabbitMQ connection configuration
RMQConnectionConfig connectionConfig =
  new RMQConnectionConfig.Builder()
    .setHost(MQ_IP)
    .setPort(MQ_PORT)
    .setUserName(MQ_UN)
    .setPassword(MQ_PW)
    .setVirtualHost("/")
    .build();

// Create RabbitMQ source
RMQSource<String> source = new RMQSource<>(
  connectionConfig, MQ_Q, true, new SimpleStringSchema());

DataStream<String> stream = env
  .addSource(source)
  .name("RabbitMQ Source");


stream.print();

Does anyone know if this is supported? If so, how can this be accomplished? We have tried using

("x-queue-type", "stream")

in ClientProperties but were unsuccessful. We'd appreciate some support, even if the advice is to take a different approach. Thanks!

I don’t think that the current Flink RabbitMQ connector supports streams

@Martijn Visser, do you know of any plans to add support for that into the existing connector? What would the advised solution be? Move to Kafka 😅?

There is https://github.com/apache/flink-connector-redis-streams/pulls but that has been quite stale for some time...

Thanks for the feedback Martijn. We are not married to RabbitMQ, so we'll rethink our choices in our tech stack.