Creating new accounts

Create an account for Kafka connector.

Procedure

  1. Select your project.
  2. Click Connectors. The list of available connectors is displayed.
  3. Click the Search icon in the search bar for Available connectors.
  4. Type Kafka and click Kafka.
  5. Specify the following details in the Added Account page to create an account that can be used to access Kafka systems.
    • In the Name field, specify the name of the account by which the Kafka server is known to Kafka developers, clients, and partners.
    • In the Description field, provide a short description of the account.
    • In the Distribution type field, choose the type of Kafka distribution that you want to connect. You can choose either Confluent Kafka(Connects to Confluent Kafka distribution) or Apache Kafka(Connects to Apache Kafka distribution).
    • In the Client type field, specify the type of account that represents whether you want to connect to a Kafka server as a producer or consumer. Possible values are -
      Producer
      Account type used to publish messages to a specific Kafka topic.
      Consumer
      Account type used to consume messages from one or more specified Kafka topics.
    • In the Bootstrap servers field, specify the list of Kafka server names with the combination of host and port, which is used for establishing the initial connection to the Kafka cluster. Use ", " (comma) to delimit multiple servers and ": " (Colon) to delimit host and port. For example - host1:port1,host2:port2.
    • In the Key serializerfield, specify the serializer type for the key. Available if the Client type is Producer.
    • In the Value serializer field, specify the serializer type for the value. Available if the Client type is Producer.
    • In the Key deserializer field, specify the deserializer type for the key. Available if the Client type is Consumer.
    • In the Value deserializer field, specify the deserializer type for the value. Available if the Client type is Consumer.
    • In the Client ID field, specify the string that identifies the client, which is passed to the Kafka server when a request is sent.
    • In the Security protocol field, specify the mode of communication and authentication mechanism that is used to exchange data that includes credentials. Possible values are -
      PLAINTEXT
      Mode of communication where the data is exchanged as plain text with no authentication. No additional configuration is required.
      SASL_PLAINTEXT
      Default. SASL is used for authentication. However, data that includes the credentials is exchanged in plain text with no encryption.
      SASL_SSL
      SASL is used for authentication and the data is SSL encrypted before exchange.
      SSL
      Authentication is not done, but data that includes credentials is SSL encrypted before the exchange. For more information about authentication mechanisms and JAAS configuration in Kafka servers, see the Kafka documentation.
      Note: When editing Apache Kafka accounts, if the Security protocol is set to -

      • PLAINTEXT, you cannot switch to other secure modes.

      • SASL_PLAINTEXT, SASL_SSL, SSL, you can switch to any of the three secure modes but not to PLAINTEXT.

    • In the Group ID field, specify a unique string that identifies the consumer group of the consumer account. Consumer in the group share the load of processing messages from topic partitions. If a consumer from a group fails, the remaining consumers in the group can take over its partitions. Available if the Client type is Consumer.
  6. Schema registry

    The schema is used to define a structured format for producing and consuming messages. The schemas are stored in the schema registry server. This functionality is applicable only for Confluent Kafka distribution. Kafka Adapter supports Avro schema where the user can select the Avro schema files from their schema registry server to produce or consume a message with a specific structure.

    Specify the details in the Schema registry fields.

    server
    The schema registry server where schemas are stored. It can follow the format such as, http://hostname:portname or https://hostname:portname or multiple schema registry servers using the comma separator.
    Note: In producer or consumer account, if Key or Value serializer or deserializer is selected as Avro, the schema registry server field is mandatory.
    credentials
    The credentials required to connect with the schema registry server.

What to do next

  • If the Client type is Producer, enter the following properties:
    Acknowledgments
    The number of acknowledgments the producer requires the lead broker to have received before considering a request as complete. This controls the acknowledgments durability of messages that are sent. For more information, see the Kafka documentation. The values that are allowed are as follows:
    • None- Message durability is high data loss risk. The producer does not wait for any acknowledgment from the broker. The message is immediately added to the socket buffer and considered as sent. It cannot be confirmed that the server receives the message. The configuration does not take effect if you retry. The offset given back for each message is always be set to -1.
    • Leader- Message durability is partial data loss risk. The lead broker writes the message to its local log but responds without awaiting for the full acknowledgement from all followers. If the leader fails immediately after acknowledging the record but before the followers replicate it, then the message is lost.
    • Replica- Message durability is no data loss risk. The lead broker waits for the full set of in-sync replicas to acknowledge the message. This helps make sure that the message is not lost when at least one in-sync replica remains alive.
    Request timeout (ms)
    Maximum amount of time in milliseconds that the client waits for the response to a request. If the response is not received before the timeout elapses, the client resends the request if necessary or fail the request if retries are exhausted. The default value is 30000.
    Compression type
    The algorithm that is used to compress the messages before they are sent to the Kafka broker. Possible values are None, GZip, Snappy, LZ4, Zstandard. The default value is None.
    Message send retries
    Number of attempts by a Kafka producer to send a message to the Kafka broker if the initial attempt fails. This helps make sure that the messages are delivered reliably in a distributed system, even when the first attempt fails due to network problems, broker unavailability, or other temporary issues. The default value is INT32_MAX, which is 2147483647.
    Batch size (bytes)
    Batch size in which the Kafka producer consolidates the messages that are sent to the same partition together before sending to the Kafka broker. Sending messages in batches improves throughput by reducing the number of requests that are sent, thus making more efficient use of network resources. The default batch size is 16384.
    TCP send buffer (bytes)
    The size of the TCP sends buffer SO_SNDBUF in bytes when sending data. If the value is set to -1, Kafka defaults to the operating system settings for the TCP send buffer size, letting the operating system manage the buffer size according to its own defaults and optimizations. The default value is 131072.
  • If the Client type is Consumer, enter the following properties:
    Enable auto commit
    Enables or disables the Kafka broker’s ability to automatically commit the consumer’s offset. When the value is set to True, auto-commit is enabled, the Kafka broker takes the responsibility for committing the consumer’s offset at the intervals that are specified in the Auto commit interval (ms)field. However, when the value is set to False, auto-commit by the Kafka broker is disabled, the Kafka connector is responsible for committing the consumer’s offsets after processing the messages. The default value is True.
    Auto commit interval (ms)
    Frequency in milliseconds that the Kafka broker auto-commits the consumer offsets to the Kafka server, if the Enable auto commit is True. The default value is 5000.
    Auto offset reset
    Defines how the Kafka consumer behaves when it needs to start reading from a topic and without a previously committed offset, or when the committed offset is invalid or out of range. Select one of the following values:
    • Earliest- Consumer starts reading from the earliest offset available in the topic, meaning it reads all messages from the beginning of the topic if there are no previously committed offsets.
    • Latest- Consumer starts reading from the latest offset, meaning it will receive only the new messages that are produced after the consumer starts. It does not process any historical messages that were produced before it started. Use this option when you care about new data and do not need to process historical messages. The field is the default setting for many Kafka clients.
    • None- Default. Consumer returns exceptions when no previous offset is found for the consumer’s group and this value is set. Use this option when it is critical to handle cases where offsets are missing or invalid explicitly and when you must help ensure that no processing occurs if no offsets are available.
    Session timeout (ms)
    Time in milliseconds that the consumer waits for a response to a request before it times out and returns an error while using the Apache Kafka’s group management facility. The default value is 45000.

  • In the Security credentials section, provide the following information.

    Truststore
    Alias of the truststore configuration. The truststore contains trusted certificates that are used to determine the trust for the remote server peer certificates. You can add a Truststore certificate by selecting the "+" icon next to the Truststore Alias drop-down list. Available if the Security protocolis SASL_SSL or SSL.
    Keystore
    Alias for the keystore configuration. You can add a Keystore certificate by selecting the " "+"" icon next to the Keystore Alias dropdown list. Available if the Security protocol is SASL_SSL or SSL.
    JAAS config
    JAAS login context parameters in the format used by JAAS configuration files. Available if the Security protocol is SASL_PLAINTEXTor SASL_SSL. For example, org.apache.kafka.common.security.plain.PlainLoginModule required username=‘KafkaConnector’ password=‘cert@XXXX’.
    Other Properties
    In the Other properties section, you can specify any additional properties that you prefer to add for the account. Type the property name and enter the value in the corresponding input text field.
    • Property name- Name of the property.
    • Property value- Value for the property.
    Note: If the Security protocol is set to SASL_PLAINTEXT or SASL_SSL, then the default value SASL mechanism is “GSSAPI”. If you want to use another SASL mechanism, configure in the Other properties field. For example, to configure the SASL mechanism to plain in the Other properties field, add the property sasl.mechanism = PLAIN.

  • Click Next. The Add account > Advanced settings page appears.

  • Configure the connection pooling manually in the Advanced settings page.

    Enable connection pooling
    Toggle the slider to right to enable configuration of the connection pooling details. Disable the connection pooling to use system-defined values. Connection pooling is disabled by default.
    Minimum pool Size
    Number of connections to create when the connection is enabled. The system maintains the specified Minimum Pool Size of connections, irrespective of whether these connections remain idle.
    Maximum pool Size
    Maximum number of connections that can exist at a time in the connection pool.
    Block timeout
    Number of milliseconds that webMethods Integration waits to obtain a connection with the IBM® Power® system before it times out and returns an error. For example, you have a pool with maximum pool size of 20. When receiving 30 simultaneous connection requests, 10 requests are queued, awaiting a connection from the pool.
    • If you set the Block timeout value to 5000, the 10 requests wait for 5 seconds to establish a connection before they time out and return an error. When the services that use the connections require 10 seconds to complete and return connections to the pool, the pending requests encounter failure and return an error stating the unavailability of the connections.
    • If you set the Block timeout value too high, you might encounter an error. If a request contains errors that delay the response, other requests will not be sent. This Block timeout must be tuned in along with the Maximum pool Size setting to accommodate such bursts in processing.
    Expire timeout
    Number of milliseconds that an idle connection can remain in the pool before it is closed and removed from the pool. The connection pool removes idle connections until the number of connections in the pool is equal to the Minimum Pool Size. The timer for an idle connection is reset when the connection is used.
    • If you set the Expire timeout value too high, the connection pool might accumulate numerous unused and idle connections. It not only consumes local memory but also ties up a connection on your backend resource. This might have an adverse effect if your resource has a limited number of connections.
    • If you set the Expire timeout value too low, the performance might degrade because of the increased activity of creating and closing connections. This Expire timeout must be tuned along with the Minimum Pool Size setting to avoid excessive opening and closing of connections during processing

  • Click Next. The Add account > Test and review page is displayed.

  • Verify the account details.
  • Click Test connection to verify the connection. A success message appears when the connectivity is successful.
  • Click Enable. A success message is displayed.
  • Click Done. You are redirected to the Connectors page and the newly created account appears in the Kafka drop-down list. You can use this account to run any Kafka custom action created under the same project.