# RabbitMQ Source Connector for Confluent Cloud The fully managed RabbitMQ Source connector for Confluent Cloud uses the AMQP protocol to communicate with RabbitMQ servers. The RabbitMQ Source connector reads data from a RabbitMQ queue (or queues) and persists the data in an Apache Kafka® topic. #### NOTE * This Quick Start is for the fully managed Confluent Cloud connector. If you are installing the connector locally for Confluent Platform, see [RabbitMQ Source Connector for Confluent Platform](https://docs.confluent.io/kafka-connectors/rabbitmq-source/current/). * If you require private networking for fully managed connectors, make sure to set up the proper networking beforehand. For more information, see [Manage Networking for Confluent Cloud Connectors](networking/internet-resource.md#clusters-connect-cloud). ## Features The RabbitMQ Source connector provides the following features: * **At least once delivery**: The connector guarantees that records are delivered at least once to the Kafka topic. * **Supports multiple tasks**: The connector supports running one or more tasks. More tasks may improve performance. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ## Limitations Be sure to review the following information. * For connector limitations, see [RabbitMQ Source Connector](limits.md#rabbitmq-source-limits) limitations. * If you plan to use one or more Single Message Transformations (SMTs), see [SMT Limitations](single-message-transforms.md#cc-single-message-transforms-limitations). #### NOTE When paused, this connector continues to consume messages from RabbitMQ until the consumer times out. These messages remain in system memory while the connector is paused. There is no data loss when the connector resumes, since messages are acknowledged after they are flushed from memory and sent to Kafka. However, if you plan to keep this connector paused for an extended time, consider removing the connector, since message will continue to accumulate in system memory. ## Quick Start Use this quick start to get up and running with the Confluent Cloud RabbitMQ Source connector. The quick start shows how to select the connector and configure it to read data from RabbitMQ queues and persist the data in an Apache Kafka® topic. Prerequisites : - Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/) cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud. - Authorized access to a RabbitMQ host server, queue, and host security details. - A topic must exist before launching the connector. - The Confluent CLI installed and configured for the cluster. See [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html). - [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). - For networking considerations, see [Networking and DNS](overview.md#connect-internet-access-resources). To use a set of public egress IP addresses, see [Public Egress IP Addresses for Confluent Cloud Connectors](static-egress-ip.md#cc-static-egress-ips).
- Kafka cluster credentials. The following lists the different ways you can provide credentials. - Enter an existing [service account](service-account.md#s3-cloud-service-account) resource ID. - Create a Confluent Cloud [service account](service-account.md#s3-cloud-service-account) for the connector. Make sure to review the ACL entries required in the [service account documentation](service-account.md#s3-cloud-service-account). Some connectors have specific ACL requirements. - Create a Confluent Cloud API key and secret. To create a key and secret, you can use [confluent api-key create](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) *or* you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector. Refer to [Cloud connector limitations](limits.md#rabbitmq-source-limits) for additional information. #### NOTE There is no `output.data.format` configuration used with this source connector. This is because this connector defaults to `ByteArrayConverter` for value and `StringConverter` for key. No other converter is useful for this connector. ### Using the Confluent Cloud Console #### Step 1: Launch your Confluent Cloud cluster To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../get-started/index.md#cloud-create-kafka-cluster). #### Step 2: Add a connector In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add connector**. #### Step 3: Select your connector Click the **RabbitMQ Source** connector card.  #### Step 4: Enter the connector details #### NOTE * Make sure you have all your [prerequisites](#cc-rabbitmq-source-prereqs) completed. * An asterisk ( \* ) designates a required entry. At the **Add RabbitMQ Source Connector** screen, complete the following: ### Topic selection Select the topic you want to send data to from the **Topics** list. To create a new topic, click **+Add new topic**. ### Kafka access 1. Select the way you want to provide **Kafka Cluster credentials**. You can choose one of the following options: - **My account**: This setting allows your connector to globally access everything that you have access to. With a user account, the connector uses an API key and secret to access the Kafka cluster. This option is not recommended for production. - **Service account**: This setting limits the access for your connector by using a [service account](service-account.md#s3-cloud-service-account). This option is recommended for production. - **Use an existing API key**: This setting allows you to specify an API key and a secret pair. You can use an existing pair or create a new one. This method is not recommended for production environments. #### NOTE Freight clusters support only service accounts for Kafka authentication. 2. Click **Continue**. ### Authentication 1. Configure the authentication properties: **How should we connect to RabbitMQ?** - **RabbitMQ host**: The RabbitMQ host server address to connect to. For example, `192.168.1.99`. Add the RabbitMQ username and password. - **RabbitMQ username**: The username to use when authenticating to RabbitMQ. - **RabbitMQ password**: The password to use when authenticating to RabbitMQ. - **RabbitMQ virtual host**: The name of the virtual host created in [RabbitMQ](https://www.rabbitmq.com/vhosts.html). - **RabbitMQ port**: RabbitMQ port to connect to. This should be set in accordance with chosen Security protocol. **Security** - **Security protocol**: The security protocol to use when connecting to RabbitMQ. Valid values are `PLAINTEXT` or `SSL`. If you select `SSL`, enter the following SSL details. - **SSL key password**: The password of the private key in the key store file. This is optional for client. - **Key Store**: The key store containing server certificate. Only required if using `SSL`. - **Key Store password**: The store password for the key store file. This is optional for a client and is only needed if key store is configured. - **Key Store type**: The file format of the key store file. This is optional for client. - **Trust Store**: The trust store containing server CA certificate. Only required if using `SSL`. - **Trust Store password**: The password for the trust store file. If a password is not set, trust store file configured will still be used, but integrity checking is disabled. Trust store password is not supported for PEM format. - **Trust Store type**: The file format of the trust store file. 2. Click **Continue**. ### Configuration **RabbitMQ details** - **Queue failure tolerance**: Controls how the connector reacts when a queue in `rabbitmq.queue` can no longer be subscribed to while the connector is running (for example, an administrator deletes the queue on the broker, permissions are revoked, or the queue is locked by another connection) and the task attempts to resubscribe on the next channel recreation. `fail-fast` (default) transitions the connector to `FAILED` on the first such queue. `skip-and-continue` skips the affected queues and continues consuming from the rest. - **RabbitMQ queue(s)**: The RabbitMQ [queue name](https://www.rabbitmq.com/queues.html#names). You can specify multiple RabbitMQ queues to read from. ### **Show advanced configurations** **Auto-restart policy** - **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its task in the event of user-actionable errors. Defaults to `true`, enabling the connector to automatically restart in case of user-actionable errors. Set this property to `false` to disable auto-restart for failed connectors. If disabled, you must manually restart the connector. **Additional Configs** - **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals: `BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and `NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value. - **Key Converter Schema ID Serializer**: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers. - **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`. - **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters. - **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry. - **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration. - **Value Converter Schema ID Serializer**: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers. **RabbitMQ details** - **Queue to topic map**: A JSON object mapping RabbitMQ queue names to Kafka topic names, for example, `{"queue-1": "topic-a", "queue-2": "topic-b"}`. Messages consumed from a mapped queue are published to its mapped topic. Queues not listed in the map, or any queue when the map is empty, fall back to the default `kafka.topic`. - **RabbitMQ batch size**: The maximum number of records that the connector batches and returns to the Kafka topic. The default property value is `1024` records. - **RabbitMQ backoff time (in milliseconds)**: The number of milliseconds (ms) to wait when no records are returned from the RabbitMQ queue. The default property value is `100` ms. **Transforms** - **Single Message Transformations**: To add a new SMT, see [Add transforms](single-message-transforms.md#cc-single-message-transforms-ui). For more information about unsupported SMTs, see [Unsupported transformations](single-message-transforms.md#cc-single-message-transforms-unsupported-transforms). See [Configuration Properties](#cc-rabbitmq-source-config-properties) for all property values and definitions. - Click **Continue**. ### Sizing Based on the number of topic partitions you select, you will be provided with a recommended number of tasks. 1. To change the number of tasks, use the Range Slider to select the desired number of tasks. 2. Click **Continue**. ### Review and Launch 1. Verify the connection details by previewing the running configuration. 2. Once you’ve validated that the properties are configured to your satisfaction, click **Launch**. The status for the connector should go from **Provisioning** to **Running**. #### Step 5: Check the Kafka topic After the connector is running, verify that messages are populating your Kafka topic. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ### Using the Confluent CLI Complete the following steps to set up and run the connector using the Confluent CLI. #### NOTE Make sure you have all your [prerequisites](#cc-rabbitmq-source-prereqs) completed. #### Step 1: List the available connectors Enter the following command to list available connectors: ```none confluent connect plugin list ``` #### Step 2: List the connector configuration properties Enter the following command to show the connector configuration properties: ```none confluent connect plugin describe ``` The command output shows the required and optional configuration properties. #### Step 3: Create the connector configuration file Create a JSON file that contains the connector configuration properties. The following example shows the required connector properties. ```none { "connector.class": "RabbitMQSource", "name": "RabbitMQSource_0", "kafka.auth.mode": "KAFKA_API_KEY", "kafka.api.key": "", "kafka.api.secret" : "", "kafka.topic" : "topic_0" "rabbitmq.host" : "192.168.1.99", "rabbitmq.username" : "", "rabbitmq.password": "", "rabbitmq.queue": "", "tasks.max" : "1" } ``` Note the following property definitions: * `"name"`: Sets a name for your new connector. * `"connector.class"`: Identifies the connector plugin name. * `"kafka.auth.mode"`: Identifies the connector authentication mode you want to use. There are two options: `SERVICE_ACCOUNT` or `KAFKA_API_KEY` (the default). To use an API key and secret, specify the configuration properties `kafka.api.key` and `kafka.api.secret`, as shown in the example configuration (above). To use a [service account](service-account.md#s3-cloud-service-account), specify the **Resource ID** in the property `kafka.service.account.id=`. To list the available service account resource IDs, use the following command: ```bash confluent iam service-account list ``` For example: ```bash confluent iam service-account list Id | Resource ID | Name | Description +---------+-------------+-------------------+------------------- 123456 | sa-l1r23m | sa-1 | Service account 1 789101 | sa-l4d56p | sa-2 | Service account 2 ``` * `"kafka.topic"`: Enter the topic name where you want data sent. A topic must exist before launching the connector. * `"rabbitmq...."`: Enter your RabbitMQ authentication and queue details. You can specify multiple `rabbitmq.queue` names to read from using a comma-separated list. * `"tasks.max"`: Enter the number of [tasks](/platform/current/connect/concepts.html#tasks) that the connector uses. The connector supports running one or more tasks. More tasks may improve performance. **SMTs**: For details about adding SMTs using the Confluent CLI, see the [Single Message Transformations](single-message-transforms.md#cc-single-message-transforms) documentation. See [Configuration Properties](#cc-rabbitmq-source-config-properties) for all property values and definitions. #### Step 4: Load the properties file and create the connector Enter the following command to load the configuration and start the connector: ```none confluent connect cluster create --config-file .json ``` For example: ```none confluent connect cluster create --config-file rabbitmq-source.json ``` Example output: ```none Created connector RabbitMQSourceConnector_0 lcc-ix4dl ``` #### Step 5: Check the connector status Enter the following command to check the connector status: ```none confluent connect plugin list ``` Example output: ```none ID | Name | Status | Type +-----------+---------------------------+---------+-------+ lcc-ix4dl | RabbitMQSourceConnector_0 | RUNNING | source ``` #### Step 6: Check the Kafka topic. After the connector is running, verify that messages are populating your Kafka topic. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ## Configuration Properties Use the following configuration properties with the fully managed connector. For self-managed connector property definitions and other details, see the connector docs in [Self-managed connectors for Confluent Platform](/platform/current/connect/kafka_connectors.html). ### How should we connect to your data? `name` : Sets a name for your connector.
* Type: string * Valid Values: A string at most 64 characters long * Importance: high ### Kafka Cluster credentials `kafka.auth.mode` : Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
* Type: string * Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY * Importance: high `kafka.api.key` : Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high `kafka.service.account.id` : The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
* Type: string * Importance: high `kafka.api.secret` : Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high ### Which topic do you want to send data to? `kafka.topic` : Identifies the topic name to write the data to.
* Type: string * Importance: high ### How should we connect to RabbitMQ? `rabbitmq.host` : RabbitMQ host to connect to.
* Type: string * Importance: high `rabbitmq.username` : Username to authenticate with RabbitMQ.
* Type: string * Importance: high `rabbitmq.password` : Password to authenticate with RabbitMQ.
* Type: password * Importance: high `rabbitmq.virtual.host` : The virtual host to use when connecting to the RabbitMQ broker.
* Type: string * Default: / * Importance: high `rabbitmq.port` : RabbitMQ port to connect to. This should be set in accordance with chosen Security protocol.
* Type: int * Default: 5672 * Valid Values: [0,…] * Importance: high ### RabbitMQ details `rabbitmq.queue` : Comma-separated list of one or more RabbitMQ queues to read from. Duplicate entries are removed automatically.
* Type: list * Importance: high `rabbitmq.queue.topic.map` : JSON object mapping RabbitMQ queue names to Kafka topic names, e.g. {“queue-1”:”topic-a”,”queue-2”:”topic-b”}. Messages consumed from a mapped queue are published to its mapped topic. Queues not listed in the map, or any queue when the map is empty, fall back to the default `kafka.topic`.
* Type: string * Default: “” * Importance: medium `rabbitmq.queue.failure.tolerance` : Controls how the connector reacts when a queue in `rabbitmq.queue` can no longer be subscribed to while the connector is running (for example, an administrator deletes the queue on the broker, permissions are revoked, or the queue is locked by another connection) and the task attempts to resubscribe on the next channel recreation. `fail-fast` (default): the connector transitions to `FAILED` on the first such queue. `skip-and-continue`: the connector skips the affected queues and continues consuming from the rest.
* Type: string * Default: fail-fast * Importance: high `batch.size` : The maximum number of records to return to Connect for each poll, if there are more than this number of records already available from RabbitMQ.
* Type: int * Default: 1024 * Valid Values: [1,…] * Importance: medium `backoff.time.ms` : The number of milliseconds to wait when no records are returned from the RabbitMQ queue.
* Type: int * Default: 100 * Valid Values: [1,…] * Importance: medium ### Security `rabbitmq.security.protocol` : The security protocol to use when connecting to RabbitMQ. Valid values are PLAINTEXT and SSL.
* Type: string * Default: PLAINTEXT * Importance: high `rabbitmq.https.ssl.key.password` : The password of the private key in the key store file. This is optional for client.
* Type: password * Importance: high `rabbitmq.https.ssl.keystorefile` : The key store containing server certificate. Only required if using SSL.
* Type: password * Default: [hidden] * Importance: high `rabbitmq.https.ssl.keystore.password` : The store password for the key store file. This is optional for a client and is only needed if key store is configured.
* Type: password * Importance: high `rabbitmq.https.ssl.keystore.type` : The file format of the key store file. This is optional for client.
* Type: string * Default: JKS * Importance: medium `rabbitmq.https.ssl.truststorefile` : The trust store containing server CA certificate. Only required if using SSL.
* Type: password * Default: [hidden] * Importance: high `rabbitmq.https.ssl.truststore.password` : The password for the trust store file. If a password is not set, trust store file configured will still be used, but integrity checking is disabled. Trust store password is not supported for PEM format.
* Type: password * Importance: high `rabbitmq.https.ssl.truststore.type` : The file format of the trust store file.
* Type: string * Default: JKS * Importance: medium ### Number of tasks for this connector `tasks.max` : Maximum number of tasks for the connector.
* Type: int * Valid Values: [1,…] * Importance: high ### Auto-restart policy `auto.restart.on.user.error` : Enable connector to automatically restart on user-actionable errors.
* Type: boolean * Default: true * Importance: medium ### Additional Configs `header.converter` : The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
* Type: string * Importance: low `producer.override.compression.type` : The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.
* Type: string * Importance: low `value.converter.allow.optional.map.keys` : Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.auto.register.schemas` : Specify if the Serializer should attempt to register the Schema.
* Type: boolean * Importance: low `value.converter.connect.meta.data` : Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.avro.schema.support` : Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.protobuf.schema.support` : Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.flatten.unions` : Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.index.for.unions` : Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.struct.for.nulls` : Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.int.for.enums` : Whether to represent enums as integers. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.latest.compatibility.strict` : Verify latest subject version is backward compatible when use.latest.version is true.
* Type: boolean * Importance: low `value.converter.object.additional.properties` : Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.optional.for.nullables` : Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.optional.for.proto2` : Whether proto2 optionals are supported. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.use.latest.version` : Use latest version of schema in subject for serialization when auto.register.schemas is false.
* Type: boolean * Importance: low `value.converter.use.optional.for.nonrequired` : Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.nullables` : Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.raw.primitives` : Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `key.converter.key.schema.id.serializer` : The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `key.converter.key.subject.name.strategy` : How to construct the subject name for key schema registration.
* Type: string * Default: TopicNameStrategy * Importance: low `value.converter.decimal.format` : Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
* Type: string * Default: BASE64 * Importance: low `value.converter.flatten.singleton.unions` : Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.reference.subject.name.strategy` : Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string * Default: DefaultReferenceSubjectNameStrategy * Importance: low `value.converter.value.schema.id.serializer` : The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `value.converter.value.subject.name.strategy` : Determines how to construct the subject name under which the value schema is registered with Schema Registry.
* Type: string * Default: TopicNameStrategy * Importance: low ## Frequently asked questions Find answers to frequently asked questions about the RabbitMQ Source connector for Confluent Cloud. ### Why is my connector failing with the `java.io.IOException` error? The `java.io.IOException occurred during connection creation: null` error typically indicates a mismatch between your port configuration and security protocol settings. Common causes include: * **Port/Protocol mismatch**: If you’re using port `5671` or `443` for AMQPS connections, you must set `rabbitmq.security.protocol` to `SSL`. The default value is `PLAINTEXT`, which only works with non-SSL ports like `5672`. * **Missing SSL configuration**: When using SSL/TLS, ensure you configure the truststore (and keystore if using mutual TLS) with valid certificates. * **Virtual host not found**: Verify that the `rabbitmq.virtual.host` value exists on your RabbitMQ server and your user has access to it. ### Why is my connector failing with authentication errors? Authentication failures can occur due to several reasons: * **Incorrect credentials**: Verify that `rabbitmq.username` and `rabbitmq.password` are correct. * **Insufficient permissions**: Ensure your RabbitMQ user has the following permissions: * Read access to the virtual host specified in `rabbitmq.virtual.host`. * Read permissions on the queues specified in `rabbitmq.queue`. * **Virtual host access**: Confirm the user has access to the specified virtual host. Use RabbitMQ management tools to verify user permissions. * **Special characters in passwords**: If your password contains special characters, ensure they are properly escaped in the configuration. ### Why does the connector fail to connect to my RabbitMQ server? Connection failures can occur due to various reasons. Follow these troubleshooting steps to diagnose and resolve the issue: 1. **Verify RabbitMQ server status**: Ensure your RabbitMQ server is running and accepting connections on the configured port. 2. **Check network connectivity**: * Confirm firewall rules allow traffic from Confluent Cloud to your RabbitMQ server. * Verify DNS resolution for the hostname specified in `rabbitmq.host`. * For private networks, ensure VPC/VNet peering or PrivateLink is properly configured. 3. **Validate configuration**: * Verify `rabbitmq.host` and `rabbitmq.port` values are correct. * Ensure `rabbitmq.security.protocol` matches your port (`SSL` for 5671, `PLAINTEXT` for 5672). * Confirm `rabbitmq.virtual.host` exists on the server. 4. **Test with RabbitMQ tools**: Use RabbitMQ management console or command-line tools to verify the server is accepting connections from the expected network path. 5. **Review connector logs**: Check the connector status and error messages in the Cloud Console. Not all detailed logs are visible in the user interface for security reasons. ### How many tasks should I configure for my connector? The RabbitMQ Source connector can run one or more tasks. The queues you configure in `rabbitmq.queue` are split across tasks, with each task supporting up to 10 queues. If `N` is the number of queues you configure, `tasks.max` must fall within the following range: ```none ceil(N / 10) ≤ tasks.max ≤ N ``` Each queue is owned by exactly one task, which preserves per-queue message ordering. The connector validates `tasks.max` against this bound at configuration time. For example, with a single queue, `tasks.max` must be exactly `1`; with 20 queues, `tasks.max` must be between `2` and `20`. #### NOTE If your workload requires multiple tasks to consume the same queue in parallel (competing-consumers behavior, which trades per-queue ordering for higher throughput on a single queue), contact [Confluent Support](https://support.confluent.io). ### Why can’t I see detailed connector logs in the Cloud Console? For security and privacy reasons, Confluent Cloud does not expose all connector logs in the UI. You can view: * **Connector status**: Overall state such as `Running`, `Failed`, and `Paused`. * **Task status**: Individual task states and high-level error messages. * **Error messages**: Summary error messages for validation and runtime failures. For detailed troubleshooting: * Review the error messages displayed in the connector status page. * Check configuration validation errors during connector creation. * Monitor connector metrics and consumer lag in the Cloud Console. * Contact [Confluent Support](https://support.confluent.io) for access to detailed logs for troubleshooting. ### How do I recover a connector from a `FAILED` state? Follow these steps to diagnose and resolve failures: 1. **Check error message**: Review the error message in the connector status page. This often indicates the specific issue (authentication, connection, configuration). 2. **Verify prerequisites**: * Confirm RabbitMQ server is accessible from Confluent Cloud. * Verify credentials and permissions are correct. * Check that virtual host and queue names are valid. * Ensure SSL/TLS configuration is correct if using secure connections. 3. **Test configuration**: Verify network connectivity using tools like `telnet` or `nc` from a host in the same network path. 4. **Restart connector**: After fixing configuration issues, restart the connector from the Cloud Console. 5. **Contact Support**: If issues persist after verifying configuration and connectivity, contact [Confluent Support](https://support.confluent.io) with: * Connector ID * Cluster ID * Error messages from the connector status page * Configuration details excluding sensitive credentials ## Next Steps For an example that shows fully managed Confluent Cloud connectors in action with Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html). This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud. [](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)