Reading an individual message from a Kafka topic

You can use the KafkaRead node to read an individual message published on a Kafka topic, by specifying the offset position of the message in the topic.

Complete the following steps to read an individual message that is published on a Kafka topic:

1. Create a message flow that contains an input node, a KafkaRead node, and an output node.
   For more information, see Creating a message flow.
2. Configure the KafkaRead node by setting the following properties:
   1. On the Basic tab, set the following properties:
      • In the Topic name property, specify the name of the Kafka topic that contains the message that you want to read.

        The topic name can be up to 255 characters in length, and can include the following characters: a-z, A-Z, 0-9, . (dot), _ (underscore), and - (dash).

      • In the Partition number property, specify the number of the Kafka partition for the topic that you want to use (valid values are between 0 and 255). The default value is 0.
      • In the Bootstrap servers property, specify the host name and port to be used for establishing the initial connection to the Kafka cluster; for example, if you are using IBM Event Streams (Kafka on IBM Cloud), specify the address of that server. You can specify multiple servers by defining a list of host:port pairs, which are separated by commas.
      • In the Message offset property, specify the offset position of the message to be read. This property defines the position in the log of messages from where the KafkaRead node reads the message. The default value is 0.

        To configure the node to retrieve the most recent message published on the partition, specify a message offset value of -1 and set the Action when message unavailable property to latest.

      • In the Action when message unavailable property, select the action to take if the message at the specified offset is not available:
        • Select no match to propagate the input message to the No match terminal. This value is the default value.
        • Select earliest to read the message with the earliest offset for the topic partition.
        • Select latest to read the message with the latest offset for the topic partition.
      • In the Client ID property, specify the client name to be used when connecting to the Kafka server.

        The client name can be up to 255 characters in length, and can include the following characters: a-z, A-Z, 0-9, . (dot), _ (underscore), and - (dash).

      • Use the Add server name suffix to client ID property to specify whether you want to suffix the client ID with the name of the integration server. This property is selected by default, and adds the integration server name to the end of the client ID, in the following format:

        -integrationServerName

        for an independent integration server, or

        integration_node_name-integration_server_name

        for an integration server that is managed by an integration node.
      • In the Timeout (s) property, specify the length of time (in seconds) in which the KafkaRead node must read the message. If the message has not been read within this timeout period, the action that is specified by the Action when message unavailable property is taken.
   2. On the Transactionality tab, set Isolation level to one of the following values:
      • Set Isolation level to one of the following values:
        • read_commited

          To receive messages, which have which were published outside of a transaction or were committed, select read_committed. If ta message is uncommitted, it is blocked until the owning transaction is committed or the transaction is rolled-back.

        • read_uncommited

          To receive messages, even if they are part of a transaction, which is not yet committed, select read_uncommitted.

   3. On the Advanced tab, set the following properties:
      • In the Properties file property, specify the path name of a kafka.properties file to be used to set Kafka client properties.

        You can use a properties file to set properties that cannot be set through the KafkaRead node properties. If the properties file specifies a property that is also set on the KafkaRead node or by a Kafka policy, the value set in the properties file takes precedence. For more information about the complete list of properties that are available, see the Apache Kafka documentation.

      • In the Header scheme property, specify how Kafka message headers are interpreted when messages are processed. Select String to interpret all Kafka headers as UTF-8 strings, or select Bytes to leave the headers in their raw binary format. The default value is String.
   4. On the Security tab, set the following properties:
      • In the Security protocol property, select the protocol to be used when communicating with the Kafka server. Valid values are:
        • PLAINTEXT
        • SSL
        • SASL_PLAINTEXT
        • SASL_SSL

        The default value for this property is PLAINTEXT.

        If you are using Event Streams, this property must be set to SASL_SSL.

        If either SASL_PLAINTEXT or SASL_SSL is selected, you must configure the user ID and password to be used to authenticate with the Kafka server by configuring the security identity, as described in Configuring security credentials for connecting to Kafka.

      • In the SSL protocol property, select the SSL protocol to be used if the Security protocol property is set to either SSL or SASL_SSL. You can select one of the following values from the editable list, or you can specify an alternative value:
        • TLSv1
        • TLSv1.1
        • TLSv1.2
        • TLSv1.3
        The default value for this property is TLSv1.2.
   5. Optional: On the Avro tab, if you want to use an Avro schema to deserialize the message that is received, complete the following steps:
      1. In the "Serialization mode" field, select Avro from the menu.
      2. In the "Policy properties" field, specify the name of the schema registry policy that contains the Avro schema.
   For more information about other properties that you can set for the KafkaRead node, see KafkaRead node. For more information about setting properties by using a policy, see Kafka policy.

   For more information about how to diagnose connection problems between IBM App Connect Enterprise and Kafka, see Resolving problems when using Kafka nodes.