Hands On: Use SMTs with a Managed Connector

In this hands on demonstration, we will be doing the following:

• Create a Datagen Source managed connector that is configured to use to SMTs
  • The Cast Value SMT will cast the orderid field as type string and the orderunits field as type int32

This exercise will be nearly the same as the Hands On: Getting Started with Kafka Connect exercise other than the addition of the two SMTs.

Let’s get started.

1. In the Confluent Cloud console, navigate into the kc-101 cluster
2. Expand Data integration and select Connectors

First things first, we’re using the Datagen connector in this exercise so let’s find and select it using the filter.

3. In the filter field, type datagen
4. Click the Datagen Source connector

We will once again generate sample data using the Orders record schema, but since we want to transform individual messages using a Single Message Transform, we will need to use the advanced configuration options.

5. Click Show advanced configuration

Let’s add our first SMT.

6. Click Add a single message transform

We could accept the default label for this transform, but it makes the configuration easier to read if we give it a name that corresponds to the SMT that will be used so let’s do this. We’re going to create an SMT to cast fields from each message, so we’ll call it “castValues”.

7. Set the value of Transform name equal to castValues

We also need to identify which SMT we want to use.

8. In the Transform type list, select org.apache.kafka.connect.transforms.Cast$Value

For this SMT, we need to enter a list of fields we want to transform and the type we want to cast them to. We do this by specifying the field name and the desired type separated by the colon. We can use any number of these using a comma-delimited list.

9. Set the value of spec equal to orderid:string, orderunits:int32

The configuration of our first SMT is complete.

Let’s now add a second SMT. This time, we’re creating an SMT to convert the timestamps of each message. Again, we’ll give it an appropriate name and fill out the required configuration parameters.

10. Click Add a single message transform.

11. Set the value of Transform name equal to convertTimestamp.

12. In the Transform type list, select org.apache.kafka.connect.transforms.TimestampConverter$Value.

13. Set the value of target.type equal to string.

This tells the SMT the resulting value should be type string.

14. Set the value of field equal to ordertime.

15. Set the value of format equal to yyyy-MM-dd.

This is the format the ordertime field will be changed to by the SMT. That completes the SMT configuration.

Let’s continue to the next step.

16. Click Continue

For this Datagen connector instance, we will once again write the output records to the orders topic.

17. Select the orders topic and click Continue

In order for our connector to communicate with our cluster, we need to provide an API key for it. You can use an existing API key and secret, or create one here, as we’re doing.

18. With Global access selected, click Generate API key & download
19. Click Continue

We will also use the default sizing for this instance of the connector.

20. Click Continue

Before we launch the connector, let’s examine its JSON configuration and identify the SMT related settings.

21. Scroll the Confluent Cloud console down so the connector JSON configuration is visible.

Notice the configuration for the two transforms is included in the connector configuration.

You could also create the connector using either the Confluent Cloud Connect API or confluent CLI and this same JSON configuration. Other than having to provide the actual value for the API key and secret, the JSON is ready to use.

Let’s now launch the connector and observe the result of the SMTs.

22. Click Launch

Now that the connector is running, let’s view messages being written to the topic and compare messages produced without the SMTs to those produced using the SMTs.

23. Navigate to Topics, select the orders topic, and select the Messages tab

Using the Jump to offset option, locate a range of records that includes the last few with the original ordertime format and the first few with the transformed ordertime format.

Note: You will need to click the pause button as soon as the display jumps to the target records to keep them in view

24. Enter values in the Jump to offset value until the ordertime format transition from original to transformed is in view and then quickly click pause

As you can see in the current view, offsets 69 and 70 have the original ordertime format and the messages written after offset 70 have the updated ordertime format. Notice also the change in data type for the orderid and orderunits fields.

Before we end the exercise, let’s delete the connector so we don’t unnecessarily deplete any Confluent Cloud promotional credits.

25. Navigate to Data integration and select Connectors
26. Select DatagenSourceConnector_0
27. Click Delete
28. Enter DatagenSourceConnector_0 in the confirmation field and click Confirm
29. Click Cluster overview

Let’s also delete the orders topic.

30. Navigate to Topics.
31. Select orders
32. Select the Configuration tab
33. Click Delete topic
34. Enter orders in the provided field to confirm the delete action and click Continue

We will not delete the kc-101 cluster at this time since we will be using it in other exercises that are part of this course.

This concludes this exercise.