Kafka Multitopic Consumer
The Kafka Multitopic Consumer origin reads data from multiple topics in an Apache Kafka cluster. The origin can use multiple threads to enable parallel processing of data. For information about supported versions, see Supported Systems and Versions in the Data Collector documentation.
When you configure a Kafka Multitopic Consumer, you configure the consumer group name and the brokers to use. You can also use a connection to configure the origin. You specify the topics to process and the number of threads to use. In Kafka, make sure that the partition assignment strategy is configured appropriately.
You can configure the origin to produce a single record when a message includes multiple objects. You can add additional Kafka configuration properties as needed. You can configure the origin to use Kafka security features. You can also configure the origin to capture Kafka message keys and store them in the record.
When processing Avro data, you can configure the Kafka Multitopic Consumer to work with the Confluent Schema Registry. The Confluent Schema Registry is a distributed storage layer for Avro schemas which uses Kafka as its underlying storage mechanism.
Kafka Multitopic Consumer includes record header attributes that enable you to use information about the record in pipeline processing.
Offset Management
The first time that a Kafka Multitopic Consumer origin identified by a consumer group receives messages from a topic, an offset entry is created for that consumer group and topic. The offset entry is created in Kafka.
- No stored offset
- When the consumer group and topic combination does not have a previously stored offset, the Kafka Multitopic Consumer origin uses the Auto Offset Reset property to determine the first message to read. You can set the origin to read messages in the topic starting from the earliest message, latest message, or a particular timestamp. The default setting is the earliest message, which results in the origin reading all existing messages in the topic.
- Previously stored offset
- When the consumer group and topic combination has a previously stored offset, the Kafka Multitopic Consumer origin receives messages starting with the next unprocessed message after the stored offset. For example, when you stop and restart the pipeline, processing resumes from the last committed offset.
Multithreaded Processing
The Kafka Multitopic Consumer origin performs parallel processing and enables the creation of a multithreaded pipeline. The Kafka Multitopic Consumer origin uses multiple concurrent threads based on the Number of Threads property and the partition assignment strategy defined in the Kafka cluster.
When performing multithreaded processing, the Kafka Multitopic Consumer origin checks the list of topics to process and creates the specified number of threads. Each thread connects to Kafka and creates a batch of data from a partition assigned by the broker based on the Kafka partition assignment strategy. Then, it passes the batch to an available pipeline runner.
A pipeline runner is a sourceless pipeline instance - an instance of the pipeline that includes all of the processors, executors, and destinations in the pipeline and handles all pipeline processing after the origin. Each pipeline runner processes one batch at a time, just like a pipeline that runs on a single thread. When the flow of data slows, the pipeline runners wait idly until they are needed, generating an empty batch at regular intervals. You can configure the Runner Idle Time pipeline property to specify the interval or to opt out of empty batch generation.
Multithreaded pipelines preserve the order of records within each batch, just like a single-threaded pipeline. But since batches are processed by different pipeline runners, the order that batches are written to destinations is not ensured.
For example, say you set the Number of Threads property to 5. When you start the pipeline, the origin creates five threads, and Data Collector creates a matching number of pipeline runners. The threads are assigned to different partitions based on the Kafka partition assignment strategy. Upon receiving data, the origin passes a batch to each of the pipeline runners for processing.
At any given moment, the five pipeline runners can each process a batch, so this multithreaded pipeline processes up to five batches at a time. When incoming data slows, the pipeline runners sit idle, available for use as soon as the data flow increases.
For more information about multithreaded pipelines, see Multithreaded Pipeline Overview. For more information about the Kafka partition assignment strategies, see the Kafka documentation.
Additional Kafka Properties
You can add custom Kafka configuration properties to the Kafka Multitopic Consumer.
When you add a custom Kafka configuration property, enter the exact property name and the value. The Kafka Multitopic Consumer does not validate the property names or values.
If custom configurations conflict with other stage properties, the stage generates an error unless you select the Override Stage Configurations check box. With the check box selected, the custom configurations override other stage properties. For example, to use a SASL mechanism other than PLAIN or GSSAPI (Kerberos), which the stage provides, add the necessary properties as custom configuration properties and select the Override Stage Configurations check box. For information about the necessary properties, see the Kafka documentation.
- auto.commit.interval.ms
- bootstrap.servers
- enable.auto.commit
- group.id
- max.poll.records
Record Header Attributes
The Kafka Multitopic Consumer origin creates record header
attributes that include information about the originating file for
the record. When the origin processes Avro data, it includes the Avro schema in
an avroSchema
record header attribute. When a Kafka message contains Kafka headers, the origin can include those headers in
the record as record header attributes.
You can use the record:attribute
or
record:attributeOrDefault
functions to access the information
in the attributes. For more information about working with record header attributes,
see Working with Header Attributes.
- avroSchema - When processing Avro data, provides the Avro schema.
- offset - The offset where the record originated.
- partition - The partition where the record originated.
- topic - The topic where the record originated.
When Data Collector uses a Kafka Java client version 0.11 or later, the origin also creates record header attributes for each message header associated with a Kafka message. The generated record header attributes have names that match the original Kafka message headers.
Kafka Security
You can configure the Kafka Multitopic Consumer origin to connect securely to Kafka through SSL/TLS, SASL, or both. For more information about the methods and details on how to configure each method, see Security in Kafka Stages.
Data Formats
The Kafka Multitopic Consumer origin processes data differently based on the data format. Kafka Multitopic Consumer can process the following types of data:
- Avro
- Generates a record for every message. Includes a
precision
andscale
field attribute for each Decimal field. - Binary
- Generates a record with a single byte array field at the root of the record.
- Datagram
- Generates a record for every message. The origin can process collectd messages, NetFlow 5 and NetFlow 9 messages, and the following types of syslog messages:
- Delimited
- Generates a record for each delimited line.
- JSON
- Generates a record for each JSON object. You can process JSON files that include multiple JSON objects or a single JSON array.
- Log
- Generates a record for every log line.
- Protobuf
- Generates a record for every protobuf message. By default, the origin assumes messages contain multiple protobuf messages.
- SDC Record
- Generates a record for every record. Use to process records generated by a Data Collector pipeline using the SDC Record data format.
- Text
- Generates a record for each line of text or for each section of text based on a custom delimiter.
- XML
- Generates records based on a user-defined delimiter element. Use an XML element directly under the root element or define a simplified XPath expression. If you do not define a delimiter element, the origin treats the XML file as a single record.