Upgrading and Migrating Apache Kafka Connector to 4.x

Upgrade Anypoint Connector for Apache Kafka (Apache Kafka Connector) to version 4.x.

Supported Upgrade Paths

From Version	To Version
3.2.x	4.x

From Version

To Version

3.2.x

4.x

Changes in This Release

Apache Kafka Connector 4.x contains a lot of changes to improve efficiency, including new operations to enable you to better understand how to use the connector.

All of the changes included in this version break backward compatibility.

This release contains the following changes:

Four new operations with new parameters were added:
- Commit
  This operation commits the offsets associated with a message or batch of messages consumed in a Message Listener (Source) or in a Batch Message Listener (Source). This operation works only with MANUAL Ack mode.
- Consume
  This operation is used for receiving messages from one or more Kafka topics. The Consume operation works similarly to the Message Listener source, so all the operations that apply to the Message Listener source also apply to this operation.
- Publish
  This operation enables sending messages to a Kafka topic. It is non-blocking, and its behavior depends on specific values of the connection parameters. This operation supports transactions, so a transaction ID is randomly generated for the Kafka Producer.
- Seek
  This operation is used for setting the current offset of the consumer for a given topic and partition to the provided offset value.
  See the New Operations table.
The Message Listener replaces the Message Consumer source.
See the Changed Sources table.
Added a Batch Message Listener source, which works in a similar way as the Message Listener with the difference that it processes the list of messages instead of a single message at a time. The message list obtained in the poll is handled by the Mule flow as a single event.
See the New Sources table for information about the new parameters.
Added different acknowledgment modes in the consumer configuration, including:
- AUTO
  Mule acknowledges the message only if the flow is finished successfully.
- MANUAL
  The user must manually acknowledge the message within the flow.
- DUPS_OK
  Same as the AUTO mode, but the commit is made asynchronously, which can result in duplicate records.
- IMMEDIATE
  Mule automatically acknowledges the message upon receipt.
There is now one plaintext connection type that supports transactions.

In version 3.x of the connector, you could add a customizable set of configuration properties related to connections. In version 4.x of the connector, this feature was removed. The connector now has a set of connection types for which you can set properties.

Changed Sources

Kafka Source	Earlier version	Parameters
Message Listener	Message consumer	Poll timeout Poll timeout time unit Acknowledgment mode Number of parallel consumers

Kafka Source

Earlier version

Parameters

Message Listener

Message consumer

Poll timeout
Poll timeout time unit
Acknowledgment mode
Number of parallel consumers

New Operations

This release contains the following new operations:

Kafka Operation	Description	Parameters
Commit	Commits the offsets associated to a message or batch of messages consumed in a Message Listener.	Consumer commit key. The consumer commit key to use to commit the message.
Consume	Enables receiving messages from one or more Kafka topics. It works similarly to the Message Listener source, so all the operations that apply to the Message Listener source also apply to this operation. [NOTE] You must specify the timeout parameter, or the flow will be blocked.	Consumption timeout Timeout time unit
Publish	Enables sending messages to a Kafka topic, and is non-blocking. Its behavior depends on specific values of the connection parameters. This operation supports transactions, which means that a transaction ID is randomly generated for the Kafka producer and handled in the connection. [NOTE] You must specify the timeout parameter, or the flow will be blocked.	Topic (The topic name) Partition (The partition name) Key Message Headers
Seek	Sets the current offset of the consumer for the given topic and partition to the provided offset value. [NOTE] You must specify the timeout parameter, or the flow will be blocked.	Topic (The topic name) Partition (The partition name) Offset (The offset to seek to)

Kafka Operation

Description

Parameters

Commit

Commits the offsets associated to a message or batch of messages consumed in a Message Listener.

Consumer commit key. The consumer commit key to use to commit the message.

Consume

Enables receiving messages from one or more Kafka topics. It works similarly to the Message Listener source, so all the operations that apply to the Message Listener source also apply to this operation.
[NOTE] You must specify the timeout parameter, or the flow will be blocked.

Consumption timeout
Timeout time unit

Publish

Enables sending messages to a Kafka topic, and is non-blocking. Its behavior depends on specific values of the connection parameters. This operation supports transactions, which means that a transaction ID is randomly generated for the Kafka producer and handled in the connection.
[NOTE] You must specify the timeout parameter, or the flow will be blocked.

Topic (The topic name)
Partition (The partition name)
Key
Message
Headers

Seek

Sets the current offset of the consumer for the given topic and partition to the provided offset value.
[NOTE] You must specify the timeout parameter, or the flow will be blocked.

Topic (The topic name)
Partition (The partition name)
Offset (The offset to seek to)

With the exception of the Commit operation, the flow will be blocked if the timeout parameter is not specified.

New Sources

Kafka Operation	Description	Parameters
Batch Message Listener	Works in the same way as the Message Listener except that it processes the list of messages instead of processing a single message at a time. The message list that was obtained in the poll is handled by a flow as a single event, so the handling of concurrency is simpler than in the simple Message Listener. In other words, a commit of the messages happens for all the messages as a whole by calling the commit on the Kafka consumer.	Poll timeout Poll timeout time unit Acknowledgment mode Number of parallel consumers

Kafka Operation

Description

Parameters

Batch Message Listener

Works in the same way as the Message Listener except that it processes the list of messages instead of processing a single message at a time. The message list that was obtained in the poll is handled by a flow as a single event, so the handling of concurrency is simpler than in the simple Message Listener. In other words, a commit of the messages happens for all the messages as a whole by calling the commit on the Kafka consumer.

Poll timeout
Poll timeout time unit
Acknowledgment mode
Number of parallel consumers

Upgrade Prerequisites

Before you perform the upgrade, you must create a backup of your files, data, and configurations in case you need to revert to the previous version.

Upgrade Steps

Follow these steps to perform the upgrade to Apache Kafka Connector 4.0.x:

In Studio, create a Mule project.
In the Mule Palette view, click Search in Exchange.
In Add Modules to Project, type 'Apache Kafka' in the search field.
In Available modules, select Apache Kafka Connector and then click Add.
Click Finish.
Anypoint Studio upgrades the connector automatically.
Verify that the kafka-connector dependency version is 4.0.0 in the pom.xml.

Post Upgrade Steps

After you install the latest version of the connector, follow these steps to complete the upgrade:

In Anypoint Studio, verify that there are no errors in the Problems or Console views.
Check the project pom.xml and verify that there are no problems.
Test the connection and verify that the operations work.

Troubleshooting

If there are problems with caching the parameters and caching the metadata, try restarting Anypoint Studio.

Reverting the Upgrade

If it is necessary to revert to the previous version of Apache Kafka Connector, in Anypoint Studio, change the kafka-connector dependency version 4.0.0 in the project’s pom.xml to the previous version.