{"revision":"c10276e953456835cdbb1f3a60d58ae1980be353","document":{"id":21089,"path":"integrate-applications/integration-hub/configure-script-consumer.md","title":"Configure a script consumer","bundle":null,"doc_type":null,"release":"australia","body":"\n# Configure a script consumer\n\nUse a script to import and process data from your Kafka environment.\n\n## Before you begin\n\n-   Role required: integration\\_hub\\_admin\n-   This consumer requires a Stream Connect subscription. For more information, see [https://www.servicenow.com/now-platform/workflow-data-fabric.html](https://www.servicenow.com/now-platform/workflow-data-fabric.html).\n-   The ServiceNow Stream Connect Installer \\[com.glide.hub.stream\\_connect.installer\\] plugin is required.\n\n## About this task\n\nTo configure a consumer, you need to create two records.\n\n1.  The consumer record, which specifies how to import and process data.\n2.  A record for the Kafka stream, which defines the stream of data to your consumer.\n\nThis task covers the consumer creation. For instructions on creating a Kafka stream, see [Create a Kafka stream](https://raw.githubusercontent.com/ServiceNow/ServiceNowDocs/australia/markdown/australia/integrate-applications/integration-hub/create-kafka-stream.md).\n\n## Procedure\n\n1.  Navigate to **All** &gt; **IntegrationHub** &gt; **Consumers** &gt; **Script Consumer**.\n\n2.  Select **New**.\n\n3.  In the form, fill in the fields.\n\n<table id=\"table_utt_jmq_hvb\"><thead><tr><th>\n\nField\n\n</th><th>\n\nDescription\n\n</th></tr></thead><tbody><tr><td>\n\nName\n\n</td><td>\n\nName of the Script consumer.\n\n</td></tr><tr><td>\n\nDelivery guarantee\n\n</td><td>\n\nIf there's a node failure, option to specify the delivery guarantee for incoming messages. Select one of the following.-   **No lost but duplicates**: All messages are delivered at least once. Some messages might be delivered more than once.\n-   **Once or not at all**: A message isn’t delivered more than once. Some messages might not be delivered at all.\n\n\n</td></tr><tr><td>\n\nSerialization format\n\n</td><td>\n\nThe serialization format for the message. Select one of the following. -   **Plain Text**: Select this option for any plain-text messages. This is the default format.\n-   **Encoded**: Select this option for messages in an Apache Avro format. Converting plain-text messages to an Avro format requires a schema. Select the schema registry in the **Schema registry** field. For more information on schemas, see [Schema management in Stream Connect](https://raw.githubusercontent.com/ServiceNow/ServiceNowDocs/australia/markdown/australia/integrate-applications/integration-hub/schema-management.md).\n\n\n</td></tr><tr><td>\n\nEvent consumer\n\n</td><td>\n\nScript to use to consume the messages received from the Kafka topic.\n\n</td></tr><tr><td>\n\nApplication\n\n</td><td>\n\nApplication scope for the Script Consumer.\n\n</td></tr><tr><td>\n\nSchema registry\n\n</td><td>\n\nRegistry for the selected schema. Select one of the following.\n\n-   **Standalone Schema Registry**\n-   **Confluent Schema Registry**\n This field appears only when the **Serialization format** is set to **Encoded**.\n\n For the Confluent Schema Registry, if the received message's schema ID isn't in the schema table, the system imports the schema dynamically, using the configured REST connection.\n\n</td></tr></tbody>\n</table>4.  Select **Save**.\n\n\n## Example\n\nThis example shows a sample script for processing messages. It includes the **utc\\_timestamp** field, a string-type field in UTC timezone, and the **metadata** field, which includes the following:\n\n-   **timestampEpoch**, data type: long\n-   **partition**, data type: int\n-   **offset**, data type: long\n-   **dataCenter**, data type: string\n\n```\n(function process(messages) {\n  for (var i = 0; i < messages.length; i++) {\n    var m = messages[i];\n\n    var headerParts = [];\n    for (var j = 0; j < m.headers.length; j++) {\n      headerParts.push('header[' + j + ']_key=' + m.headers[j].key + ' | header[' + j + ']_value=' + m.headers[j].value);\n    }\n\n    var line = [\n      'message_key='     + m.key,\n      'message_body='    + m.message,\n      'partition='       + m.metadata.partition,\n      'offset='          + m.metadata.offset,\n      'datacenter_id='   + m.metadata.datacenter_id,\n      'utc_timestamp='   + m.utc_timestamp,\n      'timestamp_epoch=' + m.metadata.timestamp_epoch\n    ].concat(headerParts).join(' | ');\n\n    gs.info('[Kafka Message] ' + line);\n  }\n})(messages);\n\n```\n\nTo convert the **utc\\_timestamp** field to GlideDateTime, see the following examples.\n\n```\n\n// Example 1\nvar timestamp = new GlideDateTime(\"2025-08-19 23:31:45\");\nconsole.log(timestamp.getValue());\n\n// Example 2 \nvar timestamp2 = new GlideDateTime();\ntimestamp2.setNumericValue(1755646305218);\nconsole.log(timestamp2.getValue());\n\n// Example 3\nvar timestamp3 = new GlideDateTime(\"1970-01-01 00:00:00\");\ntimestamp3.add(1755646305218);\nconsole.log(timestamp3.getValue());\n\n```\n\n## What to do next\n\n[Create a Kafka stream](https://raw.githubusercontent.com/ServiceNow/ServiceNowDocs/australia/markdown/australia/integrate-applications/integration-hub/create-kafka-stream.md) for this consumer. After the stream is activated, you can start receiving messages from your Kafka environment.\n\n**Parent Topic:**[Using Stream Connect for Apache Kafka](https://raw.githubusercontent.com/ServiceNow/ServiceNowDocs/australia/markdown/australia/integrate-applications/integration-hub/stream-connect-apache-kafka.md)\n\n"}}