> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# New Relic Sink Connector

> Configure and operate the New Relic Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting.

## Overview

The New Relic Sink Connector consumes records from Kafka topics, converts them into New Relic telemetry data, and sends them to New Relic. A Connector instance processes only one telemetry type. The same plugin provides three deployable concrete classes.

`com.newrelic.telemetry.events.EventsSinkConnector` converts each record into an Event. The record value must contain a string field named `eventType`. The optional `timestamp` uses a Unix timestamp in milliseconds, and the remaining supported fields become event attributes.

`com.newrelic.telemetry.logs.LogsSinkConnector` converts each record into a Log. A string field named `message` is recommended. The optional `timestamp` uses a Unix timestamp in milliseconds, and the remaining supported fields become log attributes.

`com.newrelic.telemetry.metrics.MetricsSinkConnector` converts each record into a Gauge, Count, or Summary. The record value must contain `name` and the case-sensitive `type`. `gauge` and `count` use the numeric field `value`; `summary` uses the top-level fields `aggregated_summary.count`, `aggregated_summary.sum`, `aggregated_summary.min`, and `aggregated_summary.max`; and `dimensions` provides string dimensions.

When using schemaless JSON, the record value should be converted to a Map. When using data with a schema, the record value should be converted to a Struct. With a schema, `timestamp` must be `INT64`, and the `value` for Gauge and Count must be `FLOAT64`; for Summary, `aggregated_summary.count` must be `INT32`, and the other three aggregate fields must be `FLOAT64`. Complex fields are not recursively expanded, and schemaless Event values such as booleans, objects, arrays, and `null` are not written as attributes. The Connector adds Kafka topic, partition, and offset metadata to telemetry data, but it does not automatically map the Kafka key, headers, or record timestamp to business fields.

## Prerequisites

* Prepare a New Relic account and API key that can receive telemetry data, and confirm whether the account's region is `US` or `EU`.

## License

Uses Apache License 2.0.

## Quick Start

Prepare a Connect Cluster, Kafka, a topic containing Event records, and a New Relic account, and confirm network connectivity and access permissions. For preparation and management, see [Manage connectors](../manage-connectors).

```properties theme={null}
name=new-relic-events
connector.class=com.newrelic.telemetry.events.EventsSinkConnector
topics=telemetry-events
api.key=<new-relic-api-key>
nr.region=US
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
```

Replace `<new-relic-api-key>` with the New Relic API key. If the account is in the European Union region, change `nr.region` to `EU`. The example explicitly uses schemaless JSON, so message values in the `telemetry-events` topic can use the following structure:

```json theme={null}
{
  "eventType": "OrderCreated",
  "timestamp": 1789747200000,
  "orderId": "A-1001",
  "amount": 42.5
}
```

`eventType` must be a non-empty string. If `timestamp` is omitted, the Connector uses the current time when converting the record.

## Configuration

### Instance and Telemetry Type

#### `name`

The unique name of the Connector instance.

* **Type**: `string`
* **Default**: None
* **Importance**: High
* **Valid Values / Notes**: Use a non-empty name without control characters and ensure that it is unique within the Connect Cluster.
* **Required**: Yes

#### `connector.class`

The New Relic telemetry Connector implementation class to load.

* **Type**: `string`
* **Default**: None
* **Importance**: High
* **Valid Values / Notes**: Use `com.newrelic.telemetry.events.EventsSinkConnector` for Events, `com.newrelic.telemetry.logs.LogsSinkConnector` for Logs, or `com.newrelic.telemetry.metrics.MetricsSinkConnector` for Metrics. An instance can select only one of these concrete classes.
* **Required**: Yes

### Kafka Input

#### `topics`

The list of Kafka topics to consume.

* **Type**: `list`
* **Default**: `[]`
* **Importance**: High
* **Valid Values / Notes**: Separate multiple topics with commas. Exactly one of `topics` and `topics.regex` must be configured.
* **Required**: Conditionally required

#### `topics.regex`

The Java regular expression used to dynamically match input topics.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Must be a valid Java regular expression. Exactly one of `topics` and `topics.regex` must be configured.
* **Required**: Conditionally required

### Task Parallelism

#### `tasks.max`

The maximum number of Tasks allowed to start.

* **Type**: `int`
* **Default**: `1`
* **Importance**: High
* **Valid Values / Notes**: The minimum value is `1`. Actual parallelism is limited by the number of input topic partitions. Each Task uses an independent send queue and New Relic client.

### Authentication and Region

#### `api.key`

The New Relic API key used to send telemetry data.

* **Type**: `password`
* **Default**: None
* **Importance**: High
* **Valid Values / Notes**: Must be a valid, non-empty credential. Manage it as sensitive information and do not write it to logs or commit it to a repository.
* **Required**: Yes

#### `nr.region`

The data region where the New Relic account is located.

* **Type**: `string`
* **Default**: `US`
* **Importance**: Low
* **Valid Values / Notes**: Only the case-sensitive values `US` and `EU` are supported. Explicitly configuring this value is recommended to ensure that Tasks use the intended region.

### Network Connection

#### `nr.client.timeout`

The call timeout for New Relic HTTP requests, in milliseconds.

* **Type**: `int`
* **Default**: `2000`
* **Importance**: Low
* **Valid Values / Notes**: Use a positive integer. The Connector does not validate the value range.

#### `nr.client.proxy.host`

The HTTP proxy hostname or address.

* **Type**: `string`
* **Default**: Not set (`null`)
* **Importance**: Low
* **Valid Values / Notes**: The proxy is enabled only when both `nr.client.proxy.host` and `nr.client.proxy.port` are configured.

#### `nr.client.proxy.port`

The HTTP proxy port.

* **Type**: `int`
* **Default**: Not set (`null`)
* **Importance**: Low
* **Valid Values / Notes**: The proxy is enabled only when both `nr.client.proxy.host` and `nr.client.proxy.port` are configured. Use a valid TCP port in the range `1` to `65535`.

### Batching

#### `nr.flush.max.records`

The maximum number of records in a single telemetry batch.

* **Type**: `int`
* **Default**: `1000`
* **Importance**: Low
* **Valid Values / Notes**: Use a positive integer. The Connector forms a batch immediately when this number is reached. This limit is based on the number of records, not the request size in bytes.

#### `nr.flush.max.interval.ms`

The maximum time to wait before forming a non-empty telemetry batch when the maximum record count has not been reached, in milliseconds.

* **Type**: `int`
* **Default**: `5000`
* **Importance**: Low
* **Valid Values / Notes**: Use a positive integer. A smaller value usually reduces waiting time in low-traffic scenarios but increases request frequency.

### Data Conversion and Processing

#### `key.converter`

The Converter class that deserializes Kafka message keys into Connect data.

* **Type**: `class`
* **Default**: Not set (`null`)
* **Importance**: Low
* **Valid Values / Notes**: When not set, this configuration inherits the Worker-level setting. The New Relic Connector does not map the key to telemetry business fields.

#### `value.converter`

The Converter that deserializes Kafka message values into Connect data that the Connector can process.

* **Type**: `class`
* **Default**: Not set (`null`)
* **Importance**: Low
* **Valid Values / Notes**: When not set, this configuration inherits the Worker-level setting. The converted result must be a Struct with a schema or a Map without a schema. Schemaless JSON can use `org.apache.kafka.connect.json.JsonConverter` with `value.converter.schemas.enable=false`.

#### `transforms`

The list of single message transform (SMT) aliases to apply to records in order.

* **Type**: `list`
* **Default**: `[]`
* **Importance**: Low
* **Valid Values / Notes**: Aliases must be unique. Define each transform through `transforms.<alias>.type` and the other settings for that SMT. Transforms can add or organize required fields before records enter New Relic type conversion.

## Best Practices

### Separate Instances by Telemetry Type During Initial Onboarding

**Applicable business scenario:** An organization has event, log, and metric streams in the same Kafka environment. Each data type must enter New Relic using its own field structure, and an invalid record of one type should not affect the other types. Create a separate Connector instance for each telemetry type and subscribe it to the corresponding topic.

**Configuration example:** The following instance accepts schemaless JSON logs:

```properties theme={null}
name=new-relic-logs
connector.class=com.newrelic.telemetry.logs.LogsSinkConnector
topics=telemetry-logs
api.key=<new-relic-api-key>
nr.region=US
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
```

Log message values can use the following structure:

```json theme={null}
{
  "message": "Order accepted",
  "timestamp": 1789747200000,
  "service": "checkout",
  "severity": "INFO"
}
```

The metric stream should use a separate instance and `MetricsSinkConnector`:

```properties theme={null}
name=new-relic-metrics
connector.class=com.newrelic.telemetry.metrics.MetricsSinkConnector
topics=telemetry-metrics
api.key=<new-relic-api-key>
nr.region=US
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
```

Gauge or Count message values can use the following structure:

```json theme={null}
{
  "name": "checkout.queue.depth",
  "type": "gauge",
  "value": 42.5,
  "timestamp": 1789747200000,
  "dimensions": {
    "service": "checkout",
    "region": "us-east"
  }
}
```

**Key considerations:** Do not mix the three record structures in one instance. Metric `type` recognizes only `gauge`, `count`, and `summary`; Summary uses the four top-level dotted fields `aggregated_summary.count`, `aggregated_summary.sum`, `aggregated_summary.min`, and `aggregated_summary.max`, not a nested object. Manage the topic, Converter, and message producer's data contract together to avoid mismatches between the concrete class and record structure.

### Tune Batches for Runtime Latency Requirements

**Applicable business scenario:** The Connector is sending data reliably, but telemetry from a low-traffic topic takes too long to become visible, or the number of records in each batch needs to be limited. Use both the record count and waiting time to define the batch trigger conditions.

**Configuration example:** Adjust the two batching thresholds in the Quick Start configuration:

```properties theme={null}
name=new-relic-events
connector.class=com.newrelic.telemetry.events.EventsSinkConnector
topics=telemetry-events
api.key=<new-relic-api-key>
nr.region=US
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
nr.flush.max.records=500
nr.flush.max.interval.ms=2000
```

**Key considerations:** A batch can be sent immediately when it reaches `500` records. If it does not reach that size, a non-empty batch waits at most `2000` milliseconds. These example values are not universally optimal. Monitor and tune them based on the input rate, acceptable latency, request frequency, and Worker memory. The Kafka Connect offset commit interval is not a New Relic batch trigger condition.

### Scale Tasks by Partition When Backlog Grows

**Applicable business scenario:** The input topic has multiple partitions, one Task cannot consume fast enough, and New Relic still has available ingestion capacity. Increase consumer parallelism to process the backlog.

**Configuration example:** The following configuration allows up to three Tasks to start:

```properties theme={null}
name=new-relic-events
connector.class=com.newrelic.telemetry.events.EventsSinkConnector
topics=telemetry-events
tasks.max=3
api.key=<new-relic-api-key>
nr.region=US
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
```

**Key considerations:** Effective parallelism cannot exceed the number of assignable topic partitions. Adding Tasks creates independent send queues and clients and may increase concurrent requests to New Relic. Confirm the partition count and target capacity before gradually increasing `tasks.max`. Tasks do not provide global record ordering across one another.

## Monitoring

### What to Monitor

Monitor Kafka Connect health, Connector and Task status, input and send throughput, processing and target visibility latency, consumer group offset commits and backlog, conversion or HTTP errors, SDK retries, and Worker JVM heap, GC, and thread signals. Monitor DLQ activity caused by Converter or SMT errors only when the deployment has enabled the corresponding Kafka Connect error handling.

### Import the Grafana Dashboard

Ensure that Connect metrics are available in a Grafana data source and that the collected labels match the dashboard filters; download the [Kafka Connect Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), import the JSON into Grafana, and select the corresponding data source.

## Limitations

* Events, Logs, and Metrics must use their corresponding concrete Connector classes in separate deployments. An instance cannot switch telemetry types at runtime based on record content.
* The Connector sends telemetry data asynchronously, and Kafka offsets may be committed before New Relic confirms receipt. It does not provide end-to-end exactly-once guarantees and cannot unconditionally guarantee at-least-once delivery, so records may be lost or duplicated during failure or restart windows.
* Record conversion exceptions inside the Connector are logged and skipped and do not enter the Kafka Connect DLQ. `errors.tolerance` affects only applicable errors thrown by a Converter or SMT before the record enters the Connector.
* Complex objects and arrays are not recursively expanded as general nested structures. Producers should provide supported attributes, a Metric `dimensions` map, or top-level dotted fields for Summary data, as appropriate for the selected telemetry type.
* Asynchronous retries and concurrency across multiple Tasks can change the order in which records arrive at New Relic. Do not rely on global ordering across batches or partitions.

## FAQ

### The Connector Is Running, but No Data Appears in New Relic

First confirm that `connector.class` matches the message value structure and that `api.key` and `nr.region` are correct. Check the Task logs for conversion errors, HTTP responses, and retry information. Then verify that the topic continues to receive new records, that the consumer group offset advances, and that the batch waiting time meets expectations. After correcting field types or credentials, send a new correctly formatted record to verify the result.

### The Metrics Connector Reports That the Class Cannot Be Found

The valid class name for Metrics is `com.newrelic.telemetry.metrics.MetricsSinkConnector`. Change `connector.class` to this fully qualified class name, resubmit the configuration, and confirm that the Task starts.

### Schemaless JSON Records Produce Type Conversion Errors

Confirm that `value.converter` uses `org.apache.kafka.connect.json.JsonConverter`, that `value.converter.schemas.enable=false`, and that the message value is a JSON object rather than stringified JSON. Event `eventType` and Log `message` should be strings. Metric `name` and `type` must be present, and the `value` for Gauge or Count must be parseable as a number.

### Why Did an Invalid Record Not Enter the DLQ?

New Relic type conversion occurs inside the Connector's `put` processing. This path logs and skips an invalid record instead of passing it to the Kafka Connect DLQ handler. Check the Task logs to identify the specific field and type error, then correct the record at the producer, Converter, or SMT. The DLQ can be used only for applicable Converter or SMT errors that Kafka Connect catches before invoking the Connector.

### Why Does the Runtime Display 2.3.0 When Version 2.3.3 Is Installed?

The runtime version strings for the Connector and Task in this release still return `2.3.0`, so plugin information or the New Relic `collector.version` attribute may display this value. Confirm the deployed version from the plugin artifact that is actually installed. Do not downgrade or reinstall solely because of the runtime string.