A document containing a Kafka topic, a consumer group, a partition, and an offset.

• Mapper
• CSV Formatter
• JSON Formatter
• XML Formatter

The message data with key/value parameters and metadata.

• Mapper
• Copy
• JSON Formatter
• File Writer

Required. Specify a unique name for the Snap. Modify this to be more appropriate, especially if more than one of the same Snaps is in the pipeline.

Default value: Kafka Consumer

Example: Consumer_sales_data

Default value: N/A

Default value: N/A

Default value: N/A

• If the message count is positive, the Snap terminates when it reaches the specified count, or when all available messages are read if fewer than the specified count exist.
• If the message count is negative, the Snap continues to read messages indefinitely.
• If the message count is 0, the Snap reads all currently available messages and then terminates.

Default value: -1

Select this checkbox to enable the Snap to read and process messages until the specified number in the Message Count field is reached.

If deselected, the Snap terminates when the specified number in the Message Count field is reached, or when all available messages are read if the number of message is less than the specified count.

Default status: Selected

Try decreasing this value if you encounter the error: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.

Default value: 500

Default value: 300000 (5 min.)

If the timeout expires (due to network or other issues), then the Snap polls again for more messages, unless the message count is 0, in which case it stops as expected.

Default value: 2000

If you deselect the Auto Commit check box, the Pipeline must contain one or more Acknowledge Snaps downstream to acknowledge the documents that are output by the Consumer, each of which represents an individual Kafka message, after each document is appropriately processed by the other snaps between the Consumer and Acknowledge Snaps.

Default status: Selected

Activates when you deselect the Auto Commit checkbox.

Choose the Acknowledge Mode that determines when the Snap should wait for acknowledgments from the downstream Acknowledge Snaps. The available options are:

• Wait after each record: The Snap writes a single record, then waits for that record to be acknowledged before writing another record.
• Wait after each batch of records: The Snap polls the broker to obtain a batch of records, writes all records in the batch, and then waits for each of these records to be individually acknowledged before polling again. This option has more throughput and is faster.

Default value: Wait after each record

Example: Wait after each batch of records.

Enter the maximum number of seconds to wait for a notification from the Acknowledge Snap before committing the offset.

Default value: 10

Choose an Acknowledge Timeout Policy to handle acknowledge timeout errors. Available options:

• Reprocess: This option reprocesses the unacknowledged records that are caused by acknowledge timeouts. The Snap writes an error detailing unacknowledged records to the Error view (if configured) and then commits the offset of the first unacknowledged record for reprocessing.
• Continue: This option ignores the unacknowledged records caused by acknowledge timeouts. The Snap writes an error message detailing unacknowledged records to the Error view (if configured), and then process subsequent records.

When the Consumer’s Partition is not set, an acknowledge timeout triggers a group rebalance, which can result in reprocessing in a different Consumer instance within the same group.

Default value: Reprocess

Choose the position where the Consumer Snap should begin reading messages when first initialized and executed.

The available options are:

• Beginning: Starts from the first message in the queue.
• End: Starts from the last committed offset. If none found, uses Auto Offset Reset.
• End (skip all existing messages): Starts from the end of the assigned partitions. Reads only new messages added after execution begins.
• Specify Offset: Starts from the value of the Offset field. Requires a partition number.

Default value: End

Activates when you choose Specify Offset in the Seek Type list.

Specify the Offset from which the Consumer should start consuming the message. The Offset is specific to a partition. Hence, when you specify an offset, you must specify the partition number. The offset starts from 0 for any partition. The Consumer can start consuming from any offset in the partition regardless of the messages that have been read earlier.

Default value: N/A

Choose an option for auto offset reset from the list.

If the Seek Type is End and the committed offset for the partition cannot be found, then you need to specify where to read the offset within the partition. The available options are:

• Earliest: The starting position is set to the first message in the topic.
• Latest: The starting position is set to the last offset of each partition.

This field is used only if no offsets have been committed for the current partition.

Default value: Earliest

Activates when you select the Auto Commit field, or when you set Acknowledge Mode to Wait after each batch of records.

Choose the mode to output the consumed records. The available options are:

• One output document per record: Every record received from Kafka has a corresponding output document.
• One output document per batch: Preserves the batching of records as received from Kafka. Every poll that returns a non-empty set of records results in a single output document containing the list of records as batch, batch_size and batch_index.

Default value: One output document per record

Choose the target data type for the deserialized key of each record.

• String
• JSON
• JSON_SR (requires schema registry)
• Avro (requires schema registry)
• ByteArray
• Int16, Int32, Int64
• Float32, Float64
• BigInteger, BigDecimal

Ensure that your Confluent Kafka Account's configuration includes the schema registry if you select Avro as the Key Deserializer value.

Default value: String

Choose the target data type for the deserialized value of each record.

• String
• JSON
• JSON_SR (requires schema registry)
• Avro (requires schema registry)
• ByteArray
• Int16, Int32, Int64
• Float32, Float64
• BigInteger, BigDecimal

The Avro and JSON Schema formats depend on the schema being registered in the Kafka Schema Registry. The Kafka account associated with the Snap must have the Schema Registry configured.

Default value: String

Default value: String

Select this checkbox to include a timestamp in the metadata of each record sent to the output view.

A timestamp is the number of milliseconds elapsed since midnight, January 1, 1970 UTC. All Kafka records contain a timestamp and a copy of the topic’s message.timestamp.type setting.

Default status: Not selected

Select this checkbox to allow the original document to pass through. The output includes metadata in the message regardless of this setting.

Default status: Not selected

Select one of the three modes in which the Snap executes:

• Validate & Execute: Performs limited execution and generates preview data during validation, then performs full execution during runtime.
• Execute only: Performs full execution during Pipeline runtime without generating preview data.
• Disabled: Disables the Snap and all downstream Snaps.

Default value: Validate & Execute