Customised Data Integration

Aim: Allow customised integration flows, but still allow focal control & observability over all pipelines via DI The Data Integration (DI) layer is a vital part of the Digital Integration Hub (DIH) platform. It is responsible for a wide range of data integration tasks such as ingesting data in batches or streaming data changes. This is performed in real-time from various sources and systems of record (SOR. The data then resides in the In-Memory Data Grid (IMDG), or Space, of the GigaSpaces Smart DIH platform.-Manager & SpaceDeck GigaSpaces intuitive, streamlined user interface to set up, manage and control their environment. Using SpaceDeck, users can define the tools to bring legacy System of Record (SoR) databases into the in-memory data grid that is the core of the GigaSpaces system..

How to implement: implement your own service and expose the following REST APIs REpresentational State Transfer. Application Programming Interface An API, or application programming interface, is a set of rules that define how applications or devices can connect to and communicate with each other. A REST API is an API that conforms to the design principles of the REST, or representational state transfer architectural style.:

• Start pipeline,

  POST {baseUrl}/v1/control/{pipelineId}/start?offsetStrategy={STRATEGY}&topic={topic}

• Stop pipeline,

  POST {baseUrl}/v1/control/{pipelineId}/stop

• Get status,

  GET {baseUrl}/v1/control/{pipelineId}/status

• Info,

  GET {baseUrl}/v1/control/{pipelineId}/status

See API details below.

The service implementation should have its own configuration and be responsible for knowing how to access and retrieve information from the source. Each pipeline is created by DI-Manager, which chooses the target space and keeps the configuration of that pipeline in the DI management layer.

How to Add Your Connector Service

The new service is considered a new datasource, so before creating new pipelines based on that datasource, a new datasource should be added to the system.

Using DI-Manager:

POST /api/v1/datasource/save-connection

{
				"sorName": "my-datasource",
				"dbProvider": "PLUGGABLE_CONNECTOR",
				"url": "pluggable://kafka-broker:9092"
				}
		

Create pipeline - will be done via DI-Manager / SpaceDeck as any other pipeline:

POST /api/v1/pipeline/

{
				"name": "my-pipeline",
				"sorName": "my-datasource",
				"spaceName": "my-space",
				"dataFormat": "PLUGGABLE"
				}
		

start/stop/status/info -> will be delegated to the connector service.

1. Start a Pipeline

POST {baseUrl}/v1/control/{pipelineId}/start?offsetStrategy={STRATEGY}&topic={topic}

• offsetStrategy — one of EARLIEST, LATEST, COMMITTED, OFFSET (DI Manager always sends one of these; return 400 for anything that specific implementations don't support).

• The connector can call DI Manager's GET /api/v1/pipeline/{pipelineId} to fetch the full persisted pipeline record to know the target space & other parameters.

• On success, return the plain string "started".

• On failure, return a non-2xx status. DI Manager wraps any transport error or non-2xx response into an internal exception and marks the pipeline START_FAILED in MDM.

2. Stop a Pipeline

POST {baseUrl}/v1/control/{pipelineId}/stop

• No body, no query params. Note: DI Manager's cancel flag is not forwarded to your service — you decide what "stop" means (pause vs. hard-kill) unilaterally.

• Response body ignored.

• Non-2xx → pipeline marked FAILED in MDM.

3. Get Pipeline Status

GET {baseUrl}/v1/control/{pipelineId}/status

{
				"status": "Running",
				"topic": "my-topic",
				"assignedPartitions": [0, 1, 2],
				"startTimeMs": 1718000000000,
				"consumerGroup": "my-consumer-group",
				"healthy": true,
				"lastError": null
				}
		

Field	Type	Required?	Notes
status	string	Yes	Only the literal "Running" (case-insensitive) is treated as running; anything else (including missing) maps to stopped.
healthy	boolean	No (defaults true)	status=Running + healthy=false → DI Manager reports the pipeline as FAILING, not RUNNING.
lastError	string	No	Surfaced verbatim in the DI Manager UI as the job message when healthy=false.
topic, consumerGroup, assignedPartitions	string	No	Cosmetic only — used to label synthetic "operators" in the pipeline detail UI.

our response	DI Manager job status
status=Running, healthy=true	RUNNING
status=Running, healthy=false	FAILING
status anything else	STOPPED
Endpoint unreachable / non-2xx	Internal exception, propagated up (does not silently mark STOPPED)

4. Component Info

GET {baseUrl}/v1/info

{
				"componentName": "Pluggable-Connector",
				"host": "my-host",
				"pid": 12345,
				"version": "1.2.3",
				"startTime": "2026-07-09T10:00:00Z",
				"status": "RUNNING",
				"restUrl": "http://my-host:6085"
				}
		

Error Response Convention

{"error": "<message>",	"pipelineId": "<id>"}

Register a Datasource (One-Time, Before Creating a Pipeline)

Using DI-Manager, register the connector as a data source:

POST /api/v1/datasource/save-connection

{
				"sorName": "my-datasource",
				"dbProvider": "PLUGGABLE_CONNECTOR",
				"url": "pluggable://kafka-broker:9092"
				}
		

• sorName, dbProvider, url are required; username/password are optional (only meaningful if your broker needs SASL credentials); offlineMode optional, defaults false.

• The url scheme (pluggable://...) is how DI Manager currently distinguishes this datasource type from other connector types. It's stored in DI Manager's own metadata store; nothing is sent to your connector at this point.

• Response: 200 with:

{
				"status": "SUCCESS",
				"sorName": "...",
				"dbProvider": "PLUGGABLE_CONNECTOR",
				"message": "...",
				"offlineMode": false
				}

On failure the same shape comes back with status set to FAILURE (→ 500), NOT_FOUND (→ 404), or INVALID_PARAMS (→ 400) and a message describing what went wrong — there's no separate error envelope, just a non-success status value in the normal response body.