Creating consume actions

Create a consume action.

1. In the Flow services page, click + icon to create a new flow service.
2. Provide a name and description for the flow service. The description is not mandatory.
3. On the Flow service step, type Kafka to select the Kafka connector.
4. In the Type to choose action box, click Add Custom Operation.
5. Create an account that can be used to access Kafka systems in the Account page.
   • Enter the name and description for the action.
   • Select the Kafka account of type consume.
   • Click Next.
6. Select the Consume action in the Action page and click Next.
7. In the Consume page, click "+" icon to consume messages from the topic.

   Topic

   Kafka topic from which the message must be consumed. If you choose a topic from the provided list, the message is consumed from that topic. Also, you can create a new topic and consume from it, when the Kafka broker is configured to allow the automatic creation of new topics (auto.create.topics.enable).

   Partition

   Partition where the message is stored.

   Note:

   • Subscription- If the partition is not specified, Consume action uses subscribe API allowing the Kafka broker to manage the active consumer threads in the consumer group. The Kafka broker also rebalances the partitions dynamically as and when consumers leave the group or join the group. The consumer groups when managed by Kafka broker, are deleted when no active consumers with the consumer group ID polling the topic and max.poll.interval.ms time reached. The default value for max.poll.interval.ms is 5 minutes.
   • Manual- If the partition is specified, the Consume action uses assign(partitions) API allowing the Kafka connector to manage the active consumer threads in the consumer group. The Kafka connector also takes over the responsibility of the rebalancing the partitions from the Kafka broker.

   Offset

   Name of the field that specifies the offset of the partition from where the consumer must start consuming the messages. Non-editable field. You can pass the value at runtime.

   Note: If you specify an invalid offset, the value is overwritten. It is determined based on the auto.offset.reset configuration setting. For more information see, https://kafka.apache.org/documentation.

   Note: For Apache and Confluent consumer actions, you must specify the partition at design time to view the input signature. If the partition is not specified, the input signature is not displayed, thus limiting the user's ability to pass input at runtime.

8. Provide the schema details

   Key schema

   Retrieves all the schemas registered in the schema registry server specified in the account configuration. It lists schemas only if the Key deserializer field is selected as 'Avro' in the account and the schema registry server is reachable.

   Only applicable when the account associated with the action is confluent type or belongs to confluent distribution.

   Value schema

   Retrieves all the schemas registered in the schema registry server specified in the account configuration. It lists schemas only if the Value deserializer field is selected as 'Avro' in the account and the schema registry server is reachable.

   Only applicable when the account associated with the action is confluent type or belongs to confluent distribution.

   Note: You can click the Overview button next to the Key schema and Value schema fields to view detailed schema information such as an overview of the schema, schema content, and subject versions.
9. Click Next and verify the details in the Summary page.
10. Click Show Input/Output to view the signature and click Done. You are redirected to the flow service editor's page. .
11. Click the Save icon to save the service and click the Run icon to run the service.