# WebLogic JMS Source Connector [Deprecated] for Confluent Platform The Kafka Connect WebLogic JMS Source connector is used to read messages from an [Oracle WebLogic](https://www.oracle.com/java/weblogic/) JMS Server and write them into an Apache Kafka® topic. #### NOTE Confluent Platform also includes a general [JMS Source connector for Confluent Platform](https://docs.confluent.io/kafka-connect-jms-source/current/overview.html) that uses a JNDI-based mechanism to connect to the JMS broker. The WebLogic connector also connects using JNDI, but includes special support for JMS 2.0 Shared Subscriptions. ## Features The WebLogic JMS Source connector includes the following features: - [At least once delivery]() - [JMS message types]() - [Retries and reconnection]() - [JMS 2.0 Shared Subscription support]() ### At least once delivery This connector guarantees that records are delivered at least once to the Kafka topic. If the connector restarts, there may be some duplicate records in the Kafka topic. ### JMS message types The connector currently supports [TextMessage](https://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html) and [BytesMessage](https://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html). The connector does not currently support [ObjectMessage](https://docs.oracle.com/javaee/6/api/javax/jms/ObjectMessage.html) or [StreamMessage](https://docs.oracle.com/javaee/6/api/javax/jms/StreamMessage.html). ### Retries and reconnection The connector can be configured to retry on retriable errors using the `max.retry.time` configuration property. This property sets the maximum time in milliseconds (ms) the connector will attempt to retry. The property defaults to 3600000 ms (1 hour). The connector uses exponential backoff after each retry attempt. That is, each subsequent retry attempt interval increases exponentially with jitter. The WebLogic JMS connector will attempt a retry for the following exceptions during connection/receive attempts: * `javax.jms.IllegalStateException` * `weblogic.jms.common.JMSExceptionIllegalStateException` The IllegalStateException is thrown when a receive is attempted but the server is down. The JMSException is thrown when a connection is attempted and the server is down. Note that these retries will only activate if the connector was already successfully provisioned, and then at some point later on connection/receive fails. That is, a failure to connect during provisioning will not result in retries. ### JMS 2.0 Shared Subscription Support This connector supports consuming from [JMS 2.0 shared subscriptions](https://www.oracle.com/technical-resources/articles/java/jms2messaging.html). Note that as JMS 2.0 support only begins in WebLogic 12.2.1.3, this connector only officially supports WebLogic versions later than version 12.2.1.3. Further, there is an implementation incompatibility in Oracle’s WebLogic 12.2.1.3 that limits connectors using shared subscriptions to have a maximum `tasks.max` connector configuration of 1. If you wish to leverage JMS shared subscriptions and have WebLogic 12.2.1.3, one option is to use multiple connectors each with `tasks.max` set to 1, with each connector subscribed to the same JMS topic. When connecting to WebLogic versions later than version 12.2.1.3, the connector can use more than one task during a shared subscription and each task will be a JMS shared consumer of the WebLogic JMS topic. ## Limitations The WebLogic JMS Source connector doesn’t support the Java Naming and Directory Interface (JNDI) lookup service. ## Installing the WebLogic Source Connector You can install this connector by using the [confluent connect plugin install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html) command, or by manually downloading the ZIP file. ### Prerequisites - You must install the connector on every machine where Connect will run. - If you want to install the connector using Confluent Marketplace, you must install the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This is installed by default with Confluent Enterprise. - Kafka 1.1+ or Confluent Platform 4.1+ - Oracle Weblogic v12.2.1.3+ - An installation of the Weblogic JMS client library JAR files. For help with downloading the JAR files, see the [Client Libraries](#weblogic-source-connector-client-libs) section. ### Installing the connector using Confluent Marketplace To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory and run the following command: ```bash confluent-hub install confluentinc/kafka-connect-weblogic:latest ``` You can install a specific version by replacing `latest` with a version number as shown in the following example: ```bash confluent-hub install confluentinc/kafka-connect-weblogic:13.1.0 ``` ### Installing the connector manually [Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-weblogic) for your connector and then follow the manual connector installation [instructions](https://docs.confluent.io/home/connect/install.html). ## License You can use this connector for a 30-day trial period without a license key. After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information. See [Confluent license properties](https://docs.confluent.io/kafka-connect-weblogic-source/current/source_connector_config.html#confluent-license-properties) for license properties and information about the license topic. ## Configuration Properties For a complete list of configuration properties for this connector, see [Configuration Reference for WebLogic JMS Source Connector for Confluent Platform](configuration.md#weblogic-source-connector-config). For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see [Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster). ## Client Libraries To use the Kafka Connect WebLogic JMS Source connector, you **must** download the WebLogic JMS client library JAR files. Complete the following steps to get these JAR files. #### NOTE The exact list of JAR files from the WebLogic client library may vary depending on the Java software version. 1. Follow the instructions at the WebLogic Support page [Fusion Middleware Programming Stand-alone Clients for Oracle WebLogic Server](https://docs.oracle.com/cd/E24329_01/web.1211/e24378/jms_thin_client.htm#SACLT145). The page contains an overview of the available clients and how to install them. The connector is designed to be compatible with the WebLogic JMS thin client(wlthint3client.jar) but may be compatible with others, depending on your Weblogic Server version and which JARs it provides. 2. Copy all of the required JAR files to the correct folder in `/share/`: ```text cp WL_HOME/server/lib/wlthint3client.jar /$CONFLUENT_HOME/share/confluent-hub-components/confluentinc-kafka-connect-weblogic/ ``` #### IMPORTANT These JAR files need to be copied for each of your Confluent Platform installations where you want to run this connector. 3. List the JAR files to verify that they were copied successfully. ```text ls -l /$CONFLUENT_HOME/share/confluent-hub-components/confluentinc-kafka-connect-weblogic/ ``` 4. Ensure you have the `wlthint3client.jar` file. ## Connecting to WebLogic Source Before you can use this connector, you must install the WebLogic client JARs into this connector’s installation directory. This connector connects to WebLogic using [JNDI](https://docs.oracle.com/javase/tutorial/jndi/overview/index.html) to create an instance of the JMS [ConnectionFactory](https://docs.oracle.com/javaee/7/api/javax/jms/ConnectionFactory.html) for your messaging system. See the section “Setting Up JNDI Environment Properties for the InitialContext” under Oracle’s [WebLogic JNDI](https://docs.oracle.com/cd/E13222_01/wls/docs81/jndi/jndi.html) documentation. The following example shows a typical configuration of the connector for use with [distributed mode](/platform/current/connect/concepts.html#distributed-workers): ```bash { "name": "connector1", "config": { "connector.class": "io.confluent.connect.weblogic.WeblogicSourceConnector", "kafka.topic":"MyKafkaTopicName", "java.naming.factory.initial": "weblogic.jndi.WLInitialContextFactory", "jms.destination.name":"MyWeblogicTopic", "jms.destination.type":"topic", "java.naming.provider.url": "t3://:7001/", "connection.factory.name": "com.oracle.webservices.api.jms.ConnectionFactory", "confluent.license": "", "java.naming.security.principal": "weblogic", "java.naming.security.credentials": "welcome1", "tasks.max" : "1", "jms.client.id": "id1", "jms.subscription.durable": true, "jms.subscription.shared": true, "jms.subscription.name": "sub1", "confluent.topic.bootstrap.servers":"localhost:9092" } } ``` The connector supports other [configuration options](configuration.md#weblogic-source-connector-config) not included in the example above. ## Topics This connector consumes messages from the JMS broker using the configured [message selectors](configuration.md#weblogic-source-connector-config) and writes them to a single Kafka topic. If you want to write messages to multiple topics, use a simple message transform that routes the messages based upon your criteria. ## Acknowledgement Mode The connector internally uses `CLIENT_ACKNOWLEDGE` mode to receive and acknowledge messages from the JMS broker. In this mode, acknowledging any message will acknowledge every message received (see section 6.2.10 in the [JMS Spec](https://download.oracle.com/otndocs/jcp/jms-2_0-fr-eval-spec/)). To prevent messages from being prematurely acknowledged, the connector processes only one message at time. In other words, the connector will *not* attempt to receive new messages until the last message is committed to a Kafka topic. This might compromise the throughput of the Connector, but messages will be transferred to Kafka successfully. ## Schemas The WebLogic connector produces messages with keys and values that adhere to the schemas described in the following sections. ### io.confluent.connect.jms.Key This schema is used to store the incoming MessageID on the message interface. This will ensure that when that if the same message ID arrives it will end up in the same partition. In practice this should never occur. The schema defines the following fields: | Name | Schema | Required | Default Value | Documentation | |-----------|----------|------------|-----------------|------------------------------------------------------------------------------------------------------------------------------------------------| | messageID | STRING | yes | | This field stores the value of
[Message.getJMSMessageID()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSMessageID()). | ### io.confluent.connect.jms.Value This schema is used to store the value of the JMS message. The schema defines the following fields: | Name | Schema | Required | Default Value | Documentation | |---------------|---------------------------------------------------------------------------------|------------|-----------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | messageID | STRING | yes | | This field stores the value of
[Message.getJMSMessageID()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSMessageID()). | | messageType | STRING | yes | | This field stores the type of message that was received. This corresponds
to the sub-interfaces of [Message](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html).
[BytesMessage](https://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html) =
`bytes`, [MapMessage](https://docs.oracle.com/javaee/6/api/javax/jms/MapMessage.html) = `map`,
[ObjectMessage](https://docs.oracle.com/javaee/6/api/javax/jms/ObjectMessage.html) =
`object`, [StreamMessage](https://docs.oracle.com/javaee/6/api/javax/jms/StreamMessage.html) =
`stream` and [TextMessage](https://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html) =
`text`. The corresponding field will be populated with the values from the
respective message sub-interface. | | timestamp | INT64 | yes | | Data from the [getJMSTimestamp()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSTimestamp())
method. | | deliveryMode | INT32 | yes | | This field stores the value of [Message.getJMSDeliveryMode()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSDeliveryMode()).
method. | | correlationID | STRING | no | | This field stores the value of [Message.getJMSCorrelationID()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSCorrelationID()).
method. | | replyTo | [Destination](#weblogic-source-connector-schema-destination) | no | | This schema is used to represent a JMS Destination, and is either [queue](https://docs.oracle.com/javaee/6/api/javax/jms/Queue.html) or [topic](https://docs.oracle.com/javaee/6/api/javax/jms/Topic.html). | | destination | [Destination](#weblogic-source-connector-schema-destination) | no | | This schema is used to represent a JMS Destination, and is either [queue](https://docs.oracle.com/javaee/6/api/javax/jms/Queue.html) or [topic](https://docs.oracle.com/javaee/6/api/javax/jms/Topic.html). | | redelivered | BOOLEAN | yes | | This field stores the value of [Message.getJMSRedelivered()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSRedelivered()). | | type | STRING | no | | This field stores the value of [Message.getJMSType()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSType()). | | expiration | INT64 | no | | This field stores the value of [Message.getJMSExpiration()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSExpiration()). | | priority | INT32 | no | | This field stores the value of [Message.getJMSPriority()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getJMSPriority()). | | properties | Map of STRING, [PropertyValue](#weblogic-source-connector-schema-propertyvalue) | yes | | This field stores the data from all of the properties for the Message indexed by their propertyName. | | bytes | BYTES | no | | This field stores the value from
[BytesMessage.html.readBytes(byte[])](https://docs.oracle.com/javaee/6/api/javax/jms/BytesMessage.html#readBytes(byte[])). | | map | Map of STRING, [PropertyValue](#weblogic-source-connector-schema-propertyvalue) | no | | This field stores the data from all of the map entries returned from
[MapMessage.getMapNames()](https://docs.oracle.com/javaee/6/api/javax/jms/MapMessage.html#getMapNames())
for the Message indexed by their key. | | text | STRING | no | | This field stores the value from [TextMessage.html.getText()](https://docs.oracle.com/javaee/6/api/javax/jms/TextMessage.html#getText()). | ### io.confluent.connect.jms.Destination This schema is used to represent a JMS Destination, and is either [queue](https://docs.oracle.com/javaee/6/api/javax/jms/Queue.html) or [topic](https://docs.oracle.com/javaee/6/api/javax/jms/Topic.html). The schema defines the following fields: | Name | Schema | Required | Default Value | Documentation | |-----------------|----------|------------|-----------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | destinationType | STRING | yes | | The type of JMS Destination, and either `queue` or `topic`. | | name | STRING | yes | | The name of the destination. This will be the value of
[Queue.getQueueName()](https://docs.oracle.com/javaee/6/api/javax/jms/Queue.html#getQueueName())
or [Topic.getTopicName()](https://docs.oracle.com/javaee/6/api/javax/jms/Topic.html#getTopicName()). | ### io.confluent.connect.jms.PropertyValue This schema is used to store the data that is found in the properties of the message. To ensure that the proper type mappings are preserved field `propertyType` stores the value type for the field. The corresponding field in the schema will contain the data for the property. This ensures that the data is retrievable as the type returned by [Message.getObjectProperty()](https://docs.oracle.com/javaee/6/api/javax/jms/Message.html#getObjectProperty(java.lang.String)). The schema defines the following fields: | Name | Schema | Required | Default Value | Documentation | |--------------|----------|------------|-----------------|-----------------------------------------------------------------------------------------------------------------------------------------| | propertyType | STRING | yes | | The Java type of the property on the Message. One of `boolean`, `byte`,
`short`, `integer`, `long`, `float`, `double`, or `string`. | | boolean | BOOLEAN | no | | The value stored as a boolean. Null unless `propertyType` is set to `boolean`. | | byte | INT8 | no | | The value stored as a byte. Null unless `propertyType` is set to `byte`. | | short | INT16 | no | | The value stored as a short. Null unless `propertyType` is set to `short`. | | integer | INT32 | no | | The value stored as a integer. Null unless `propertyType` is set to `integer`. | | long | INT64 | no | | The value stored as a long. Null unless `propertyType` is set to `long`. | | float | FLOAT32 | no | | The value stored as a float. Null unless `propertyType` is set to `float`. | | double | FLOAT64 | no | | The value stored as a double. Null unless `propertyType` is set to `double`. | | string | STRING | no | | The value stored as a string. Null unless `propertyType` is set to `string`. |