Skip to main content

​
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.
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:
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:
Log message values can use the following structure:
The metric stream should use a separate instance and MetricsSinkConnector:
Gauge or Count message values can use the following structure:
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:
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:
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, 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.