Redis Streams logo Redis Streams agent

Target Redis Streams

Prerequisites

To have Gluesync append to Redis Streams you will need:

  • A Redis server that accepts XADD on the stream keys this agent resolves;

  • Network reachability to the host and port (and to every cluster seed when additionalHosts is set);

  • Either disableAuth set to true, or a password and, for an ACL user, a username;

  • Permission to run XADD, PING and INFO server on that server;

  • When TLS is on and peer verification stays enabled, a trust store the JVM already has, a certificatePath, or a certificate uploaded for the agent.

Streams do not have to exist beforehand. This agent creates a stream by the first XADD. It does not run DDL.

Setup via Web UI

Connection fields come from RedisStreamsNodeInformationDao.hostCredentialsFields. Where the code leaves the display name empty, the label below is the same wording the other target pages use for that key, and the key is in parentheses.

  • Protocol (protocol): Fixed option Vendor SDK. Default Vendor SDK.

  • Hostname / IP Address (host): Hostname or IP address of the Redis node. Required.

  • Port (port): Optional, defaults to 6379.

  • Username (username): ACL username. Optional. Blank means password-only authentication when a password is set. See Security.

  • Password (password): Password for the ACL user, or the legacy password when username is blank.

  • Database index (databaseName): Logical database index sent with Lettuce withDatabase on a standalone connection. Default 0. A blank value is database 0. Any other value must be a non-negative integer. Ignored in cluster mode (the URI is built with withDatabase false).

  • Disable auth (disableAuth): Optional, defaults to false. When true, no username or password is sent.

  • Enable Tls (enableTls): Optional, defaults to false.

  • Trust server certificate (trustServerCertificate): Optional, defaults to false. When true and TLS is on, peer verification is NONE.

  • Certificate path (certificatePath): Optional. Trust-store path used when TLS is on and trustServerCertificate is false. When this is blank, the agent looks up a store uploaded for the agent. When that is also missing, it keeps the JVM default trust store.

  • Certificates password (certificatePassword): Optional. Password of the trust store above.

  • Additional hosts (additionalHosts): Optional. When the list is not empty the agent connects with RedisClusterClient. Each entry has host and an optional port (a missing port reuses the main port). The main host and port are the first seed.

The entity structure fields are:

  • Key namespace (optional) (schema): Optional scope. Used as {scope} in the stream key.

  • Stream (table): Collection name. Used as {collection} in the stream key.

Custom host credentials

These are RedisStreamsCustomHostCredentials.

  • Command timeout (seconds) (commandTimeoutSeconds): Optional, defaults to 60. Applied as the Lettuce command timeout and as the connection timeout. PING and each XADD batch wait up to this long.

  • Client name (clientName): Optional, defaults to gluesync-redis-streams. Sent with withClientName when it is not blank.

Specific configuration

These are RedisStreamsSpecificConfiguration. Names are matched to the enum values ignoring case. A value that is not one of the listed options fails on the next write with the allowed names. Invalid combinations fail on the next write as well: jsonFieldName blank, SENTINEL with an empty nullSentinel, MAXLEN with trimMaxLen not greater than zero, MINID with trimMinIdRetentionSeconds not greater than zero, or pipelineBatchSize not greater than zero.

  • Stream key prefix (streamKeyPrefix): Optional, default empty. Prepended in the default stream key. Also the value of {prefix} in a template.

  • Stream key template ({prefix}, {scope}, {collection}, {entity}; blank = auto) (streamKeyTemplate): Optional, default empty. See Stream key.

  • Payload layout (payloadLayout): FLAT (default) or JSON.

  • JSON field name (JSON layout) (jsonFieldName): Optional, default payload. The single field name used when the layout is JSON. Must not be blank.

  • Metadata field prefix (flat layout) (metadataFieldPrefix): Optional, default _. Prefixed to metadata field names in the FLAT layout. Not trimmed.

  • Null values (flat layout) (nullValueMode): OMIT_AND_LIST (default) or SENTINEL.

  • Null sentinel (flat layout, SENTINEL mode) (nullSentinel): Optional, default NULL. Written for a null column when the mode is SENTINEL. Must not be empty in that mode.

  • Binary encoding (binaryEncoding): HEX (default) or BASE64. HEX is lowercase hex (HexFormat). BASE64 is standard Base64.

  • Include before image (includeBeforeImage): Default false. When true, an update entry includes the old column values and a delete entry includes the row image the source sent. Inserts and truncate markers do not get a before image.

  • Emit truncate marker events (emitMarkerEvents): Default true. When true, truncate and snapshot deletion append one truncate marker. When false, the agent logs a warning and does not append (RedisStreamsEntryPublisher.publishMarker).

  • Trim strategy (trimStrategy): NONE (default), MAXLEN or MINID. Passed as XADD arguments (RedisStreamsTrimming). NONE sends XADD with no trim clause.

  • Trim MAXLEN (entries kept per stream) (trimMaxLen): Optional, default 0. Used only when the strategy is MAXLEN, and then it must be greater than zero. Keeps that many newest entries (MAXLEN).

  • Trim MINID retention (seconds) (trimMinIdRetentionSeconds): Optional, default 0. Used only when the strategy is MINID, and then it must be greater than zero. The agent converts seconds to milliseconds and sends MINID of <now minus that many milliseconds>-0, and never below 0-0. now is the clock at append time.

  • Approximate trimming (~) (trimApproximate): Default true. true sends approximate trimming (~); false sends exact trimming.

  • Pipeline batch size (XADD per round trip) (pipelineBatchSize): Default 1000. Must be greater than zero. Number of XADD commands issued before the agent waits for that batch (RedisStreamsWriter).

sourceChangeRetention (Source change retention in hours) and technicalFields (Technical fields) are on the same class but are marked for a source agent. This connector does not use them when it writes.

There are no table-level properties (RedisStreamsTableConfiguration is empty).

Per-entity configuration

These are RedisStreamsEntityConfiguration. They are not agent-wide credentials.

  • Stream key override (streamKey): Optional, default empty. When not blank, this exact string is the stream key and the template and the default key are not used.

  • Pipeline batch size override (0 = agent default) (pipelineBatchSize): Optional, default 0. A value greater than zero replaces the agent pipelineBatchSize for that entity. 0 (and any value that is not greater than zero) keeps the agent default.

Setup via Rest APIs

Here following an example of calling the Core Hub’s Rest API via curl to setup the connection for this Agent.

Connect the agent

Standalone connection. Leave username empty for password-only authentication, or set disableAuth to true to send neither.

curl --location --request PUT 'http://core-hub-ip-address/pipelines/{pipelineId}/agents/{agentId}/config/credentials' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
    "hostCredentials": {
        "connectionName": "myAgentNickName",
        "protocol": "Vendor SDK",
        "host": "host-address",
        "port": 6379,
        "username": "",
        "password": "",
        "databaseName": "0",
        "disableAuth": false,
        "enableTls": false,
        "trustServerCertificate": false,
        "certificatePath": null,
        "certificatePassword": null
    },
    "customHostCredentials": {
        "commandTimeoutSeconds": 60,
        "clientName": "gluesync-redis-streams"
    }
}'

Cluster mode is the same call with additionalHosts set. The database index is not applied in that mode.

"additionalHosts": [
    { "host": "10.0.0.2", "port": 6379 },
    { "host": "10.0.0.3", "port": null }
]

Setup specific configuration

The following example shows how to apply the agent-specific configurations via Rest API.

curl --location --request PUT 'http://core-hub-ip-address/pipelines/{pipelineId}/agents/{agentId}/config/specific' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
  "configuration": {
      "streamKeyPrefix": "",
      "streamKeyTemplate": "",
      "payloadLayout": "FLAT",
      "jsonFieldName": "payload",
      "metadataFieldPrefix": "_",
      "nullValueMode": "OMIT_AND_LIST",
      "nullSentinel": "__NULL__",
      "binaryEncoding": "HEX",
      "includeBeforeImage": false,
      "emitMarkerEvents": true,
      "trimStrategy": "NONE",
      "trimMaxLen": 0,
      "trimMinIdRetentionSeconds": 0,
      "trimApproximate": true,
      "pipelineBatchSize": 1000
  }
}'

Stream key

StreamKeyResolver names the stream for each entity. Highest precedence first:

  1. The entity Stream key override (streamKey), used verbatim when it is not blank.

  2. Stream key template (streamKeyTemplate), when it is not blank. The agent replaces these placeholders and nothing else: {prefix} (the stream key prefix), {scope} (the entity scope, trimmed), {collection} (the entity collection, trimmed), {entity} (the entity full name: scope.collection when the scope is not blank, otherwise collection).

  3. Otherwise <streamKeyPrefix>:<scope>:<collection>, joining with : and skipping any part that is blank. With an empty prefix and an empty scope the key is the collection name.

The result must not be blank. If it is, the write fails and asks for a collection, a stream key prefix, or a per-entity stream key.

This key is only the Redis stream name. It is not the document id carried inside the entry. See Document id.

Data types

The agent accepts all 16 Gluesync data types (RedisStreamsDataTypeMapping). Each one has a single target type of its own, so there is nothing to choose in the Fields Editor. The target type is only a label: it never creates anything on Redis. Values reach the writer untouched and RedisStreamsValueCodec turns them into text.

Gluesync type Target type Flat layout JSON layout

STRING

string

As is.

String.

BOOLEAN

boolean

true or false.

Boolean.

SHORT, INT, LONG

short, int, long

Decimal text.

Number.

FLOAT, DOUBLE

float, double

Decimal text.

Number. NaN and infinities are strings.

BIG_DECIMAL

bigdecimal

Plain decimal text, trailing zeros of the source scale removed (23.50 becomes 23.5).

String with the same text, so no digit is lost.

BYTE_ARRAY

binary

Lowercase hex, or Base64 when Binary encoding is BASE64.

String with the same text.

LOCAL_DATE

date

ISO-8601.

String.

LOCAL_TIME

time

ISO-8601.

String.

LOCAL_DATE_TIME

datetime

ISO-8601.

String.

OFFSET_TIME

offset_time

ISO-8601.

String.

OFFSET_DATE_TIME

offset_date_time

ISO-8601.

String.

ARRAY, MAP

array, map

JSON text.

JSON array or object, elements following this same table.

What is appended

RedisStreamsEntryPublisher builds one entry per change and appends it with XADD. The three target DAOs (single table, multi table, NoSQL entity) all delegate there. The operation word is the source operation (StreamEntryFactory): insert, update, delete or truncate. A snapshot row keeps the operation the source gave that row and sets snapshot to true.

Nothing in this agent deletes or trims a key inside the payload. The only removal of old entries is the optional MAXLEN or MINID clause on XADD itself. XADD does not set the entry id; Redis assigns it. The agent does not publish a second, empty entry after a delete.

Operation What is appended

Insert

op insert. Row columns in the after image. No before image.

Update

op update. Row columns in the after image. Before image only when Include before image is true (the old value of each column).

Delete

op delete. The after image is omitted. In the flat layout the key columns are still written as row fields. Before image, when enabled, is the row the source sent for that delete. The stream entry is not removed.

Truncate

One marker, op truncate, snapshot false, no document id, no key columns, no row. Only when Emit truncate marker events is true.

Snapshot deletion

The same marker with snapshot true. Only when marker events are enabled.

Delivery is at-least-once. XADD never fails as a duplicate key (isDuplicateKeyFailure is false, insertConflictSupport is NOT_SUPPORTED). If a batch fails after Redis has already accepted some of its commands, those entries stay and a later retry appends them again (rejectedBatchLeavesNothingWritten is false). RedisCommandTimeoutException and RedisConnectionException are retried in place (RedisStreamsTargetAgentService). Other errors are not. The command timeout is commandTimeoutSeconds.

Batches are issued in order on one connection and awaited one batch after another, including in cluster mode, because one stream key is owned by one node (RedisStreamsWriter).

On connect the agent calls PING and requires the reply PONG.

Document id

The _key / key value is AffectedData.toId, not the stream key.

When the entity has no custom key mapping, the id is the key-column values joined by :: (the default separator), with a prefix of collection:: or, when the scope is not blank, scope::collection::. A custom key mapping can replace the separator, the prefix, the suffix and which columns are the keys. A change that does not carry a mapped key column is refused. A key column that is present but null is kept as an empty component.

Markers do not have a document id.

Flat layout

Default. Metadata fields are written first, each name prefixed with metadataFieldPrefix (default _). Then one field per row column. Field order is the order below; Lettuce sends that order.

Field (default prefix) When it is present

_op

Always. insert, update, delete or truncate.

_snapshot

Always. true or false.

_entity

Always. scope.collection, or collection when the scope is blank.

_key

When the document id is present. Omitted on markers.

_keys

JSON object of the key columns, when there is at least one. Omitted on markers.

_ts_ms

Always. Agent clock when the entry was built, epoch milliseconds.

_source_ts_ms

Only when the source transaction id has a timestamp greater than zero. Epoch milliseconds.

_transaction

Serialized transaction id. For a snapshot row that has none of its own, the snapshot’s starting transaction id. Omitted when there is none (markers, and a delete, which is published without the fallback id).

_before

Only when Include before image is true and this operation has a before image. JSON object.

row columns

Insert and update: every column of the after image. Delete: the key columns only. A null column is omitted and its name is listed in _nulls (OMIT_AND_LIST), or written as nullSentinel (SENTINEL).

_nulls

JSON array of column names that were null, only in OMIT_AND_LIST and only when that list is not empty.

A column whose name is already one of those metadata field names (including _nulls) fails the write. Change metadataFieldPrefix in that case.

Every field value is text. Numbers and booleans are their decimal or true/false text. BIG_DECIMAL is plain decimal text with trailing zeros of the source scale removed (23.50 becomes 23.5); an integer keeps every digit and is not written with an exponent. Dates and times are ISO-8601 (LOCAL_DATE, LOCAL_TIME, LOCAL_DATE_TIME, OFFSET_TIME, OFFSET_DATE_TIME). A byte array is lowercase hex or Base64, following Binary encoding. Arrays and maps are JSON text. A BigInteger is its decimal text. Any other JVM value is toString().

JSON layout

One field, named by jsonFieldName (default payload). The value is one JSON document. Nulls are JSON null. The keys, in order, are op, snapshot, entity, key, ts_ms, source_ts_ms, transaction, keys, after, before. after is null for a delete and for a marker. before is null unless a before image was built.

In that document, booleans and finite integers, longs, shorts, bytes, floats and doubles stay JSON booleans and numbers. BIG_DECIMAL and BigInteger are JSON strings (same plain text as the flat layout). Non-finite floats and doubles are strings. Temporals and binaries are strings with the same text as the flat layout. Nested arrays and maps use the same rules.

Example

Defaults, collection customers, no scope, no key-mapping override, one key column id = 7, full_name = Ada, age null, not a snapshot, flat layout. The document id and the transaction text come from the source row; the numbers below are only an illustration of the field names.

XADD customers * \
  _op insert \
  _snapshot false \
  _entity customers \
  _key customers::7 \
  _keys {"id":7} \
  _ts_ms 1700000000000 \
  id 7 \
  full_name Ada \
  _nulls ["age"]

_source_ts_ms and _transaction are added only when the source row has them. The same row with payloadLayout JSON is a single payload field whose document contains "after":{"id":7,"full_name":"Ada","age":null} and "before":null.