This page describes the Lenses API references.
Lenses offers a set of REST and WebSocket APIs to help you provision and manage the application, make use of its capabilities, and query data via SQL.
This page describes how to authentication against Lenses APIs.
All requests must be authenticated using an HTTP Header x-kafka-lenses-token:myToken. You can obtain the token via the following login API or you can use a service account.
All REST APIs are protected via role-based authentication that is either BASIC or LDAP based, depending on how Lenses security has been set up. In order to be able to use the APIs, you will need to first authenticate via an appropriate user, then receive an access token and use that token for any subsequent request.
POST/api/login
Headers
Body
Response
To run the example below, we recommend installing the
API reference documentation for Lenses.
Content-Type
application/json
user
string
Name of the user
password
string
Password of the user
"a1f44cb8-0f37-4b96-828c-57bbd8d4934b"UNAUTHORIZED # login and receive the access token
HOST="http://localhost:3030"
curl -X POST -H "Content-Type:application/json" \
-d '{"user":"admin", "password":"admin"}' \
${HOST}/api/login -s --compressed -w '\n'"a1f44cb8-0f37-4b96-828c-57bbd8d4934b"A WebSocket allows the client to request data using SQL as input.
Use the following URL to open a WebSocket connection. Depending on deployment, use ws (not-secure) or wss (secure) connection:
Once the connection is established, the server expects the client to send a JSON message with the following structure:
The client should wait for incoming messages and stop when EOF is received. The server sends JSON-encoded messages, the payload structure is:
Since there are multiple types of messages the server sends, the type attribute determines the data structure for the data payload. Below is the list of possible type values:
The data the attribute is null. The client should ignore these messages.
In this case, the payload structure is:
When requested the payload format is:
A possible payload can be:
The payload format is:
The payload format is:
The payload format is:
The payload format is:
Given the type RECORD the payloads contains the data returned to the client, and it is dependent on the SQL input. The payload format is:
For a topic which contains nested data for both key and value.
The payload format is:
If the Kafka message key is null the response would be:
Projecting the key will not return the key attribute:
produces the following output:
Aggregating data will not return the key and metadata attributes:
WS | WSS ${LENSES_URL}/api/ws/v2/sql/execute{
"token": String,
"sql": String,
"live": Optional[Boolean],
"stats": Optional[Int]
}Millis interval to receive query stats. If not provided the stats information is not sent
optional[int]
Reserved.
PAGE_END
Reserved.
END
The message signals the end of the execution. The server will close the socket after this message is sent.
BADRECORD
The message contains information about the Kafka message which cannot be read. For example, if the topic expects Avro and the payload is not a valid Avro.
SENTINEL
The message contains information about the execution termination reason when query thresholds are reached.
token
Login token for lenses.
long
sql
The SQL query used to fetch data
string
live
If set, enables partial results for aggregation queries.
optional[boolean]
HEARTBEAT
To keep the connection alive, the server injects an empty message.
RECORD
The message represents a data record.
ERROR
The message contains information about an error that occurred.
STATS
The message contains information about the current execution.Only activated if the stats field in the request is present.
METADATA
The message contains a list of metadata fields each record message will contain. For example: offset, partition,timestamp, __keysize, __valueSize, valueSchemaId.
data.key
Returns the underlying Kafka message key value. Only applied when * projection is used. When a key projection is used (i.e. _key.MMSI), the value is returned as data.value.MMSI.
data.data
Will contain the output generated by the SQL projections. If * is used for the SQL projection, it will return the Kafka message value structure.
data.metadata
Returned when no aggregations are involved. It contains the Kafka message partition, offset, byte size information
data.rownum
Reserved optional long value.
stats
SCHEMA
{
"type": <Enumeration>,
"data": <Dependent on the type field>
}{
"type": "HEARTBEAT"
}{
"type": "HEARTBEAT",
"data": String
}{
"type": "HEARTBEAT",
"data": {
id: String,
bytesRead: Long,
sources: Map[UUID, String],
sourceConfigs: Map[UUID, Map[String, Any]],
sourcesStats: Map[UUID, Map[Int, Map[String, Long]]],
badRecordsTotal: Long,
recordsSkipped: Long,
results: Long,
recordsScanned: Long,
duration: Long,
startedAt: Long,
}{
"type": "STATS",
"data": {
"id": "6a29ef99-3914-4640-b0dc-ec4a07c7ffc5",
"bytesRead": 51972426,
"sources": {
"ad1afc73-3dec-41e7-b679-aa35c55e3328": "cc_payments"
},
"sourceConfigs": {
"ad1afc73-3dec-41e7-b679-aa35c55e3328": {
"live.aggs": true,
"max.idle.time": 20000,
"format.timestamp": true,
"show.bad.records": true,
"kafka.offset.timeout": 10000,
"max.query.time": 3600000,
"max.size": 209715200
}
},
"sourcesStats": {
"ad1afc73-3dec-41e7-b679-aa35c55e3328": {
"0": {
"begin": 0,
"offsetBound": 541950,
"bytesRead": 51972426,
"recordsScanned": 541951,
"end": 541950
}
}
},
"badRecordsTotal": 0,
"recordsSkipped": 0,
"results": 119,
"recordsScanned": 541951,
"duration": 18196,
"startedAt": 1678901355694
}
}{
"type": "METADATA",
"data": {
"fields": Array[String]
}
}{
"type": "END"
}{
"type": "BADRECORD",
"data": {
"metadata": {
"offset": Optional[Long],
"partition": Optional[Long],
"timestamp": Optional[Long],
"table": String,
"sourceId": String
}
}
}{
"type": "SENTINEL",
"data": {
"event": String,
"reason": {
"type": String,
"timeout": Optional[Long],
"source": String,
"sourceId": String,
"amount":Long
}
}
}{
"type": "RECORD",
"data": {
"fields": <dynamic>,
"value": <dynamic>,
"metadata": <dynamic>,
"rownum": Optional[Long]
}
}SELECT * FROM sea_vessel_position_reports{
"type": "RECORD",
"data": {
"key": {
"MMSI": 219005662
},
"value": {
"Type": 1,
"Repeat": 0,
"MMSI": 219005662,
"Speed": 0.1,
"Accuracy": false,
"Longitude": 12.570543333333333,
"Latitude": 55.85040166666667,
"location": "55.850402,12.570543",
"Course": 177.9,
"Heading": 511,
"Second": 20,
"RAIM": false,
"Radio": 33577,
"Status": 3,
"Turn": -128.0,
"Maneuver": 0,
"Timestamp": 1491318144456507320
},
"metadata": {
"offset": 71,
"partition": 1,
"timestamp": "2023-03-13T11:42:24.267Z",
"__keysize": 10,
"__valuesize": 79,
"keySchemaId": 4,
"valueSchemaId": 5
},
"rownum": 71
}
}{
"type": "RECORD",
"data": {
"value": {
"Type": 1,
"Repeat": 0,
"MMSI": 219005662,
"Speed": 0.1,
"Accuracy": false,
"Longitude": 12.570543333333333,
"Latitude": 55.85040166666667,
"location": "55.850402,12.570543",
"Course": 177.9,
"Heading": 511,
"Second": 20,
"RAIM": false,
"Radio": 33577,
"Status": 3,
"Turn": -128.0,
"Maneuver": 0,
"Timestamp": 1491318144456507320
},
"metadata": {
"offset": 71,
"partition": 1,
"timestamp": "2023-03-13T11:42:24.267Z",
"__keysize": 10,
"__valuesize": 79,
"keySchemaId": 4,
"valueSchemaId": 5
},
"rownum": 71
}
}SELECT _key.MMSI FROM sea_vessel_position_reports{
"type": "RECORD",
"data": {
"value": {
"MMSI": 265527470
},
"metadata": {
"offset": 86,
"partition": 1,
"timestamp": "2023-03-13T11:42:29.107Z",
"__keysize": 10,
"__valuesize": 79,
"keySchemaId": 4,
"valueSchemaId": 5
},
"rownum": 86
}
}SELECT count(*), currency
FROM cc_payments
GROUP BY currency{
"type": "RECORD",
"data": {
"value": {
"count": 20066,
"currency": "NOR"
},
"rownum": 90
},
"rowId": "fc767952-97e5-46cf-868c-7cbfbcde21fb",
"statementIndex": null
}It will update the connections state and validate the configuration. If the validation fails, the state will not be updated.
It will only validate the request, not applying any actual change to the system.
falseIt will try to connect to the configured service as part of the validation step.
trueSuccessfully updated connection state
Bad request
Attached file(s) needed for establishing the connection. The name of each file part is used as a reference in the manifest.
{
"updated": [
"text"
],
"created": [
"text"
],
"deleted": [
"text"
]
}GET /api/v1/state HTTP/1.1
Host:
Accept: */*
{
"license": {
"maxBrokers": 1,
"expiry": 1,
"clientId": "text",
"isRespected": true,
"status": "Valid",
"message": "text"
},
"connections": {
"kafka": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"confluentSchemaRegistry": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"elasticSearch": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"pagerDuty": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"datadog": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"slack": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"alertManager": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"webhook": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"aws": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"connect": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"awsGlueSchemaRegistry": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"zookeeper": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"postgres": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"splunk": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
],
"kerberos": [
{
"name": "text",
"version": 1,
"tags": [
"text"
]
}
]
}
}PUT /api/v1/state/license HTTP/1.1
Host:
Content-Type: application/json
Accept: */*
Content-Length: 65
{
"source": "text",
"clientId": "text",
"details": "text",
"key": "text"
}{
"maxBrokers": 1,
"expiry": 1,
"clientId": "text",
"isRespected": true,
"status": "Valid",
"message": "text",
"settings": {
"security": {
"root": {
"enabled": true
},
"basic": {
"enabled": true
},
"ldap": {
"enabled": true
},
"kerberos": {
"enabled": true
},
"custom": {
"enabled": true
},
"sso": {
"enabled": true
},
"serviceAccount": {
"enabled": true,
"restriction": {
"name": "None"
}
}
},
"sql": {
"streaming": {
"enabled": true,
"restriction": {
"name": "None"
}
},
"sql": {
"enabled": true
}
},
"kafkaSettings": {
"acls": true,
"quotas": true,
"consumerOffsetManagement": true
},
"audit": {
"enabled": true,
"integration": true
},
"connections": {
"enabled": true
},
"application": {
"topology": true,
"connectorsOnKubernetes": true
},
"approval": {
"enabled": true
},
"alerts": {
"enabled": true,
"rules": {
"name": "None"
},
"integration": {
"enabled": true,
"channels": [
"text"
],
"max": {
"name": "None"
}
}
},
"data": {
"masking": true,
"customSerde": true,
"sla": true,
"namespace": {
"enabled": true,
"max": {
"name": "None"
}
}
},
"backup": {
"enabled": true
}
},
"currentTime": 1
}PUT /api/v1/state/connections HTTP/1.1
Host:
Content-Type: multipart/form-data
Accept: */*
Content-Length: 4971
{
"provisioning": {
"kafka": [
{
"name": "kafka",
"version": 1,
"tags": [
"text"
],
"configuration": {
"protocol": {
"value": "PLAINTEXT"
},
"sslKeystore": {
"file": "text"
},
"sslKeystorePassword": {
"value": "text"
},
"sslKeyPassword": {
"value": "text"
},
"sslTruststorePassword": {
"value": "text"
},
"sslTruststore": {
"file": "text"
},
"saslJaasConfig": {
"value": "text"
},
"keytab": {
"file": "text"
},
"kafkaBootstrapServers": {
"value": [
"text"
]
},
"saslMechanism": {
"value": "text"
},
"metricsPort": {
"value": 1
},
"metricsUsername": {
"value": "text"
},
"metricsPassword": {
"value": "text"
},
"metricsSsl": {
"value": true
},
"metricsHttpSuffix": {
"value": "text"
},
"metricsHttpTimeout": {
"value": 1
},
"metricsType": {
"value": "AWS"
},
"additionalProperties": {
"value": {}
},
"metricsCustomUrlMappings": {
"value": {}
},
"metricsCustomPortMappings": {
"value": {}
}
}
}
],
"confluentSchemaRegistry": [
{
"name": "schema-registry",
"version": 1,
"tags": [
"text"
],
"configuration": {
"sslKeystore": {
"file": "text"
},
"sslKeystorePassword": {
"value": "password"
},
"sslKeyPassword": {
"value": "password"
},
"sslTruststorePassword": {
"value": "password"
},
"sslTruststore": {
"file": "text"
},
"schemaRegistryUrls": {
"value": [
"text"
]
},
"basicAuthCredentialsSource": {
"value": "password"
},
"basicAuthUserInfo": {
"value": "password"
},
"metricsType": {
"value": "JMX"
},
"metricsSsl": {
"value": true
},
"metricsUsername": {
"value": "text"
},
"metricsPassword": {
"value": "password"
},
"metricsPort": {
"value": 1
},
"additionalProperties": {
"value": {}
},
"metricsCustomUrlMappings": {
"value": {}
},
"metricsCustomPortMappings": {
"value": {}
},
"metricsHttpSuffix": {
"value": "text"
},
"metricsHttpTimeout": {
"value": 1
},
"username": {
"value": "text"
},
"password": {
"value": "password"
},
"hardDelete": {
"value": true
}
}
}
],
"elasticSearch": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"user": {
"value": "text"
},
"password": {
"value": "password"
},
"nodes": {
"value": [
"text"
]
}
}
}
],
"pagerDuty": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"integrationKey": {
"value": "text"
}
}
}
],
"datadog": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"site": {
"value": "EU"
},
"apiKey": {
"value": "text"
},
"applicationKey": {
"value": "text"
}
}
}
],
"slack": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"webhookUrl": {
"value": "text"
}
}
}
],
"alertManager": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"endpoints": {
"value": [
"text"
]
}
}
}
],
"webhook": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"host": {
"value": "text"
},
"port": {
"value": 1
},
"useHttps": {
"value": true
},
"creds": {
"value": [
"text"
]
}
}
}
],
"aws": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"authMode": {
"value": "Credentials Chain"
},
"accessKeyId": {
"value": "text"
},
"secretAccessKey": {
"value": "text"
},
"region": {
"value": "text"
},
"sessionToken": {
"value": "text"
}
}
}
],
"connect": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"workers": {
"value": [
"text"
]
},
"username": {
"value": "text"
},
"password": {
"value": "text"
},
"metricsSsl": {
"value": true
},
"metricsUsername": {
"value": "text"
},
"metricsPassword": {
"value": "text"
},
"metricsType": {
"value": "JMX"
},
"metricsPort": {
"value": 1
},
"aes256Key": {
"value": "text"
},
"sslAlgorithm": {
"value": "text"
},
"sslKeystore": {
"file": "text"
},
"sslKeystorePassword": {
"value": "text"
},
"sslKeyPassword": {
"value": "text"
},
"sslTruststorePassword": {
"value": "text"
},
"sslTruststore": {
"file": "text"
},
"metricsCustomUrlMappings": {
"value": {
"ANY_ADDITIONAL_PROPERTY": "text"
}
},
"metricsCustomPortMappings": {
"value": {
"ANY_ADDITIONAL_PROPERTY": 1
}
},
"metricsHttpSuffix": {
"value": "text"
},
"metricsHttpTimeout": {
"value": 1
}
}
}
],
"awsGlueSchemaRegistry": [
{
"name": "schema-registry",
"version": 1,
"tags": [
"text"
],
"configuration": {
"authMode": {
"reference": "text"
},
"accessKeyId": {
"reference": "text"
},
"secretAccessKey": {
"reference": "text"
},
"sessionToken": {
"value": "text"
},
"glueRegistryArn": {
"value": "text"
},
"glueRegistryCacheTtl": {
"value": 1
},
"glueRegistryCacheSize": {
"value": 1
},
"schemaRegistryFlavour": {
"value": "text"
},
"glueRegistryDefaultCompatibility": {
"value": "BACKWARD"
}
}
}
],
"zookeeper": [
{
"name": "zookeeper",
"version": 1,
"tags": [
"text"
],
"configuration": {
"zookeeperUrls": {
"value": [
"text"
]
},
"zookeeperChrootPath": {
"value": "text"
},
"zookeeperSessionTimeout": {
"value": 1
},
"zookeeperConnectionTimeout": {
"value": 1
},
"metricsType": {
"value": "JMX"
},
"metricsPort": {
"value": 1
},
"metricsUsername": {
"value": "text"
},
"metricsPassword": {
"value": "text"
},
"metricsSsl": {
"value": true
},
"metricsHttpSuffix": {
"value": "text"
},
"metricsHttpTimeout": {
"value": 1
},
"metricsCustomUrlMappings": {
"value": {}
},
"metricsCustomPortMappings": {
"value": {}
}
}
}
],
"postgres": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"host": {
"value": "text"
},
"port": {
"value": 1
},
"database": {
"value": "text"
},
"username": {
"value": "text"
},
"password": {
"value": "text"
},
"sslMode": {
"value": "allow"
}
}
}
],
"splunk": [
{
"name": "text",
"version": 1,
"tags": [
"text"
],
"configuration": {
"host": {
"value": "text"
},
"port": {
"value": 1
},
"useHttps": {
"value": true
},
"insecure": {
"value": true
},
"token": {
"value": "text"
}
}
}
],
"kerberos": [
{
"name": "kerberos",
"version": 1,
"tags": [
"text"
],
"configuration": {
"kerberosKrb5": {
"file": "text"
}
}
}
]
},
"attachedFile": "binary"
}