{"revision":"c10276e953456835cdbb1f3a60d58ae1980be353","document":{"id":836,"path":"api-reference/server-api-reference/KafkaProducerAPIScoped.md","title":"KafkaProducerAPI - Scoped","bundle":null,"doc_type":null,"release":"australia","body":"\n# KafkaProducerAPI- Scoped\n\nThe KafkaProducerAPI provides a method to produce messages from server-side scripts to an Apache Kafka topic.\n\nTo access this API, the ServiceNow IntegrationHub Action Step - Kafka Producer plugin \\(`com.glide.hub.action_step.kafka`\\) must be activated. This API runs in the `sn_ih_kafka` namespace.\n\nUse the send\\(\\) method to deliver a message to a Kafka topic. Before calling send\\(\\), a [topic alias](https://raw.githubusercontent.com/ServiceNow/ServiceNowDocs/australia/markdown/integrate-applications/manage-topic-alias.md) record for the target Kafka topic must exist in the Topic Alias \\[sys\\_sc\\_topic\\_alias\\] table.\n\nPermission for sending messages to a given topic is enforced by access control on the topic alias record.\n\nFor more information about working with Kafka topics, see [Using Stream Connect for Apache Kafka](https://raw.githubusercontent.com/ServiceNow/ServiceNowDocs/australia/markdown/integrate-applications/stream-connect-apache-kafka.md).\n\n**Parent Topic:**[Server API reference](https://raw.githubusercontent.com/ServiceNow/ServiceNowDocs/australia/markdown/australia/api-reference/server-api-reference/api-server.md)\n\n## KafkaProducerAPI - KafkaProducerAPI\\(\\)\n\nInstantiates a KafkaProducerAPI.\n\n|Name|Type|Description|\n|----|----|-----------|\n|None| ||\n\nCreates an instance of KafkaProducerAPI.\n\n```\nvar kafkaProducerAPI = new sn_ih_kafka.KafkaProducerAPI();\n```\n\n## KafkaProducerAPI - send\\(Object params\\)\n\nSends a message to a Kafka topic, optionally running a script include if an asynchronous send fails.\n\nBefore calling this method for asynchronous message delivery, create a script include to handle delivery failures. When the script include executes, an array of error objects \\(**callbackErrors**\\) is passed to the callback function. Each error object represents one message delivery failure.\n\n\n
\n\nName\n\n\n\nType\n\n\n\nDescription\n\n
\n\ncontext\n\n\n\nString\n\n\n\nThe value passed as **params.context** in the original send\\(\\) call for the failed message. Data type: String\n\n
\n\nerrorMessage\n\n\n\nString\n\n\n\nError message describing the reason for failure. Data type: String\n\n
\n\nerrorType\n\n\n\nString\n\n\n\nThe fully qualified class name for the exception. Data type: String\n\n
\n\nproducerId\n\n\n\nString\n\n\n\nSys\\_id of the producer event associated with the failed send.Table: Kafka Producers \\[sys\\_kafka\\_producer\\]\n\nData type: String\n\n
\n
\n\nName\n\n\n\nType\n\n\n\nDescription\n\n
\n\nparams\n\n\n\nObject\n\n\n\nObject containing parameters to use for sending the message.```\n{\n context: \"String\",\n headers: {Object},\n isSync: Boolean,\n key: \"String\",\n message: \"String\",\n onErrorScriptIncludeId: \"String\",\n topicSysId: \"String\" \n}\n```\n\n
\n\nparams.context\n\n\n\nString\n\n\n\nOptional. Caller-supplied context string to pass to the script include if an asynchronous send fails. Used to identify the message that triggered the failure.\n\n
\n\nparams.headers\n\n\n\nObject\n\n\n\nOptional. Key-value pairs for message headers. Any headers supported by Apache Kafka are valid. For example:```\n{\n \"source\": \"order_update_br\",\n \"correlation_id\": \"8a1c3f2b1b3a10\"\n}\n```\n\n
\n\nparams.isSync\n\n\n\nBoolean\n\n\n\nOptional. Flag that indicates whether to send the message synchronously. Possible values:- true: Send synchronously and block for the result.\n- false: Send asynchronously. An error callback can be specified in **onErrorScriptIncludeId** to execute when an asynchronous send fails.\n\nDefault: false\n\n
\n\nparams.key\n\n\n\nString\n\n\n\nOptional. Message key for partitioning in Apache Kafka. Messages with the same key are sent to the same partition, ensuring ordering.Default: Empty string\n\n
\n\nparams.message\n\n\n\nString\n\n\n\nContent of the message to send to the topic.\n\n
\n\nparams.onErrorScriptIncludeId\n\n\n\nString\n\n\n\nOptional. Sys\\_id of the script include to run if an asynchronous send fails. Default: None\n\nTable: Script Includes \\[sys\\_script\\_include\\]\n\n
\n\nparams.schemaSysId\n\n\n\nString\n\n\n\nOptional. Sys\\_id of a registered schema record to use for encoding the message.Default: The message is sent as plain text.\n\nTable: Stream Connect Schemas \\[stream\\_connect\\_schema\\]\n\n
\n\nparams.topicSysId\n\n\n\nString\n\n\n\nSys\\_id of the target Kafka topic alias record.Table: Topic Alias \\[sys\\_sc\\_topic\\_alias\\]\n\n
\n
\n\nType\n\n\n\nDescription\n\n
\n\nObject\n\n\n\nMap containing delivery result information.For synchronous delivery, the method returns after the message is delivered. For asynchronous delivery, the method returns immediately. The delivery may later succeed or fail. Asynchronous delivery failure can be handled using the provided **onErrorScriptIncludeId**.\n\n```\n{\n delivery: \"String\",\n offset: Number,\n partition: Number\n}\n```\n\n
\n\n<Object>.delivery\n\n\n\nDelivery type.Possible values:\n\n- `async`: The message will be sent asynchronously.\n- `sync`: The message was sent synchronously.\n\nData type: String\n\n
\n\n<Object>.offset\n\n\n\nUnique ID assigned to the message within the topic partition. Only returned for `sync` delivery.Data type: Number\n\n
\n\n<Object>.partition\n\n\n\nPartition that the message was delivered to. Only returned for `sync` delivery.Data type: Number\n\n
This business rule sends an order-updated event asynchronously, routing any delivery failure to the example `KafkaSendFailureHandler` script include. Note that `KafkaSendFailureHandler` is shown only as an example; you must create your own script include to handle delivery failures.\n\n```\nvar kafkaProducerAPI = new sn_ih_kafka.KafkaProducerAPI();\n\nvar response = kafkaProducerAPI.send({\n topicSysId: '8a1c3f2b1b3a1010e7cfc51b1ce0bc42', // sys_id for 'orders.updated'\n key: current.getUniqueValue(),\n message: JSON.stringify({\n order_id: current.getUniqueValue(),\n status: current.getValue('status')\n }),\n isSync: false,\n headers: { source: 'order_update_br' },\n onErrorScriptIncludeId: 'b2e4a1c3f2b1a1010e7cfc51b1ce0c99', // sys_id of KafkaSendFailureHandler\n context: current.getUniqueValue()\n});\n\ngs.info('Kafka send submitted, delivery mode: ' + response.delivery);\n// Output:\n// Kafka send submitted, delivery mode: async\n```\n\nThis script include reads **callbackErrors** and logs each delivery failure.\n\n```\n// Script include: KafkaSendFailureHandler\n(function callback(callbackErrors) {\n\n gs.info(\"callback size: \", callbackErrors.length);\n\n for (var i = 0; i < callbackErrors.length; i++) {\n gs.info(\"callback \" + callbackErrors[i].errorMessage + \" \" +\n callbackErrors[i].errorType + \" \" +\n JSON.stringify(callbackErrors[i].context) + \" \" +\n callbackErrors[i].producerId);\n }\n\n})(callbackErrors);\n```\n\n"}}