<a id="pagerduty-sink-connector"></a>

# PagerDuty Sink Connector for Confluent Platform [Deprecated]

#### IMPORTANT
- This connector is deprecated and will reach its end of life (EOL) on the Confluent Platform 7.9 end of support (EOS) date.
  Confluent recommends migrating to [HTTP Sink V1 connector](https://docs.confluent.io/kafka-connectors/http/current/overview.html) before the EOL date.
  For more information, see [Connector support lifecycle policy](https://docs.confluent.io/platform/current/connect/supported.html#support-policy-for-self-managed-connectors).
- <!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

  Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
  [Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
  page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
  are no longer available for download.

The Kafka Connect PagerDuty Sink connector is used to read records from an
Apache Kafka® topic and create [PagerDuty incidents](https://support.pagerduty.com/docs/incidents).

<a id="pagerduty-sink-connector-feature"></a>

## Features

The PagerDuty Sink connector offers the following features:

- [At least once delivery](#pager-duty-sink-at-least-once-delivery)
- [Dead Letter Queue](#pager-duty-sink-dead-letter-queue)
- [Multiple tasks](#pager-duty-sink-multiple-tasks)
- [Automatic retries](#pager-duty-sink-automatic-retries)
- [HTTPS proxy](#pager-duty-sink-https-proxy)
- [CSFLE (Client-side Field level encryption)](#pager-duty-sink-csfle-sm)

<a id="pager-duty-sink-at-least-once-delivery"></a>

### At least once delivery

The connector creates a PagerDuty incident for each record in Kafka topic.
However, duplicates are still possible to occur when failure, rescheduling or
reconfiguration happens. This semantics is followed when `behavior.on.error`
is set to `fail` mode. In case of `log` and `ignore` modes, the connector
promises at-most semantics.

<a id="pager-duty-sink-dead-letter-queue"></a>

### Dead Letter Queue

This connector supports the Dead Letter Queue (DLQ) functionality. For
information about accessing and using the DLQ, see [Confluent Platform
Dead Letter Queue](/platform/current/connect/concepts.html#dead-letter-queue).

<a id="pager-duty-sink-multiple-tasks"></a>

### Multiple tasks

The PagerDuty Sink connector supports running one or more tasks. You can
specify the number of tasks in the `tasks.max` configuration parameter. This
can lead to performance gains when multiple files need to be parsed.

<a id="pager-duty-sink-automatic-retries"></a>

### Automatic retries

The PagerDuty Sink connector may experience network failures while connecting to
the PagerDuty endpoint. The connector will automatically retry with exponential
backoff to create incidents. The property `pagerduty.max.retry.time.seconds`
controls the maximum time until which the connector will retry creating the
incidents.

<a id="pager-duty-sink-https-proxy"></a>

### HTTPS proxy

The connector can connect to PagerDuty using an HTTPS proxy server.

<a id="pager-duty-sink-csfle-sm"></a>

### CSFLE (Client-side Field level encryption)

This connector supports the CSFLE functionality. For more information, see [Manage CSFLE](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

<a id="pagerduty-sink-connector-license"></a>

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

For license properties, see [Confluent Platform license](connector_config.md#pagerduty-sink-license-config). For
information about the license topic, see
[License topic configuration](connector_config.md#pagerduty-sink-license-topic-configuration).

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for PagerDuty Sink Connector for Confluent Platform](connector_config.md#pagerduty-sink-connector-config).

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

<a id="pagerduty-sink-connector-install"></a>

## Install PagerDuty Sink Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later.
- Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later.
- Java 1.8.
- **Team tier** [PagerDuty account](https://www.pagerduty.com/sign-up/?type=team).
- [PagerDuty Service ID](#pagerduty-service-id).
- An installation of the latest (`latest`) connector version.

  To install the `latest` connector version, navigate to your Confluent Platform
  installation directory and run the following command:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-pagerduty:latest
  ```

  You can install a specific version by replacing `latest` with a version
  number as shown in the following example:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-pagerduty:1.0.7
  ```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-pagerduty) for your
connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

<a id="pagerduty-sink-connector-quickstart"></a>

## Quick Start

The quick start guide uses PagerDuty Sink connector to consume records from a
Kafka topic and create incidents in PagerDuty.

<a id="pagerduty-service-id"></a>

### Get service ID

If you have a service already created in PagerDuty, you can find the Service ID
by navigating to **Configuration** > **Services**. Select your service. The
Service ID is in the URL bar. For example:
`https://connector-test.pagerduty.com/service-directory/<serviceId>`.

You can add a service in PagerDuty. When you add a service, select **Use our API
directly** as the Integration type. After the service is added, you will be
redirected to the URL where you can copy the Service ID.

![PagerDuty API selection screen](images/cp-pagerduty-use-api.png)

PagerDuty integration setting

### Start Confluent

Start the Confluent services using the following [Confluent
CLI](https://docs.confluent.io/confluent-cli/current/index.html) command:

```bash
confluent local start
```

#### IMPORTANT
Do not use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.

### Property-based example

Create a configuration file `pagerduty-sink.properties` with the following
content. This file should be placed inside the Confluent Platform installation directory. This
configuration is used typically along with [standalone
workers](/platform/current/connect/concepts.html#standalone-workers).

```text
name=pagerduty-sink-connector

topics=incidents
connector.class=io.confluent.connect.pagerduty.PagerDutySinkConnector
tasks.max=1

pagerduty.api.key=****
behavior.on.error=fail

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081

confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1
confluent.license=

reporter.bootstrap.servers=localhost:9092
reporter.result.topic.replication.factor=1
reporter.error.topic.replication.factor=1
```

#### NOTE
For details about using this connector with Kafka Connect Reporter, see
[Connect Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).

Run the connector with this configuration.

```bash
confluent local load pagerduty-sink-connector --config pagerduty-sink.properties
```

The output should resemble:

```json
 {
    "name":"pagerduty-sink-connector",
    "config":{
        "topics":"incidents",
        "tasks.max":"1",
        "connector.class":"io.confluent.connect.pagerduty.PagerDutySinkConnector",

        "pagerduty.api.key":"****",
        "behavior.on.error":"fail",

        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter":"io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url":"http://localhost:8081",

        "confluent.topic.bootstrap.servers":"localhost:9092",
        "confluent.topic.replication.factor":"1",
        "reporter.bootstrap.servers": "localhost:9092",
        "reporter.result.topic.replication.factor":"1",
        "reporter.error.topic.replication.factor":"1"

        "name":"pagerduty-sink-connector"
     },
    "tasks":[
      {
        "connector":"pagerduty-sink-connector",
        "task":0
      }
     ],
     "type":"sink"
}
```

#### NOTE
For details about using this connector with Kafka Connect Reporter, see
[Connect Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter).

Confirm that the connector is in a `RUNNING` state.

```bash
confluent local status pagerduty-sink-connector
```

The output should resemble:

```bash
{
   "name":"pagerduty-sink-connector",
   "connector":{
      "state":"RUNNING",
      "worker_id":"127.0.1.1:8083"
   },
   "tasks":[
      {
         "id":0,
         "state":"RUNNING",
         "worker_id":"127.0.1.1:8083"
      }
   ],
   "type":"sink"
}
```

### REST-based example

Use this setting with [distributed
workers](/platform/current/connect/concepts.html#distributed-workers). Write the following JSON to
`config.json`, configure all of the required values, and use the following
command to post the configuration to one of the distributed connect workers.
Check here for more information about the Kafka Connect [Kafka
Connect REST Interface](/platform/current/connect/references/restapi.html)

```json
{
  "name" : "pagerduty-sink-connector",
  "config" : {
    "topics":"incidents",
    "connector.class":"io.confluent.connect.pagerduty.PagerDutySinkConnector",
    "tasks.max" : "1",

    "pagerduty.api.key":"****",
    "behavior.on.error":"fail",

    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter":"io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url":"http://localhost:8081",

    "confluent.topic.bootstrap.servers":"localhost:9092",
    "confluent.topic.replication.factor":"1",
    "confluent.license":" Omit to enable trial mode ",

    "reporter.bootstrap.servers": "localhost:9092",
    "reporter.result.topic.replication.factor":"1",
    "reporter.error.topic.replication.factor":"1"
  }
}
```

#### NOTE
Change the `confluent.topic.bootstrap.servers` property to include your
broker address(es) and change the `confluent.topic.replication.factor` to
`3` for staging or production use.

Use curl to post a configuration to one of the Kafka Connect workers. Change
`http://localhost:8083/` to the endpoint of one of your Kafka Connect
worker(s).

```bash
curl -sS -X POST -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors
```

Use the following command to update the configuration of existing connector.

```bash
curl -s -X PUT -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors/pagerduty-sink-connector/config
```

Confirm that the connector is in a `RUNNING` state by running the following command:

```bash
curl http://localhost:8083/connectors/pagerduty-sink-connector/status | jq
```

The output should resemble:

```bash
{
   "name":"pagerduty-sink-connector",
   "connector":{
      "state":"RUNNING",
      "worker_id":"127.0.1.1:8083"
   },
   "tasks":[
      {
         "id":0,
         "state":"RUNNING",
         "worker_id":"127.0.1.1:8083"
      }
   ],
   "type":"sink"
}
```

To produce Avro data to Kafka topic: `incidents`, use the following command.

```bash
./bin/kafka-avro-console-producer --broker-list localhost:9092 --topic incidents --property value.schema='{"type":"record","name":"details","fields":[{"name":"fromEmail","type":"string"}, {"name":"serviceId","type":"string"},{"name":"incidentTitle","type":"string"}]}'
```

While the console is waiting for the input, use the following three records and
paste each of them on the console.

```bash
{"fromEmail":"user1@abc.com", "serviceId":"<your-service-id>", "incidentTitle":"Incident Title x 0"}
{"fromEmail":"user2@abc.com", "serviceId":"<your-service-id>", "incidentTitle":"Incident Title x 1"}
{"fromEmail":"user3@abc.com", "serviceId":"<your-service-id>", "incidentTitle":"Incident Title x 2"}
```

#### NOTE
The `fromEmail` in the records should be of registered PagerDuty user to
allow creation of incidents

Finally, check the PagerDuty incident dashboard to see the newly created
incidents.

<a id="pagerduty-record-schema"></a>

## Record schema

The PagerDuty Sink connector expects the value of records in Kafka topic to be
either of type JSON String or Avro. In either case, the value should adhere to
the following conditions:

1. Must include `fromEmail`, `serviceId` and `incidentTitle` fields in the value of the Kafka record.
2. The `body`, `urgency`, `escalationPolicy` and  `priority` fields are optional.

Following is the value schema for Pagerduty incident:

```json
{
   "name":"PagerdutyValueSchema",
   "type":"record",
   "fields":[
      {
         "name":"fromEmail",
         "type":"string"
      },
      {
         "name":"serviceId",
         "type":"string"
      },
      {
         "name":"incidentTitle",
         "type":"string"
      },
      {
         "name":"priorityId",
         "type":"string",
         "isOptional":true
      },
      {
         "name":"urgency",
         "type":"string",
         "isOptional":true
      },
      {
         "name":"bodyDetails",
         "type":"string",
         "isOptional":true
      },
      {
         "name":"escalationPolicyId",
         "type":"string",
         "isOptional":true
      }
   ]
}
```