Kafka-GigaSpaces Connector Features

The Pluggable Connector is a flexible, YAML-based data integration solution that seamlessly connects Apache Kafka Apache Kafka is a distributed event store and stream-processing platform. Apache Kafka is a distributed publish-subscribe messaging system. A message is any kind of information that is sent from a producer (application that sends the messages) to a consumer (application that receives the messages). Producers write their messages or data to Kafka topics. These topics are divided into partitions that function like logs. Each message is written to a partition and has a unique offset, or identifier. Consumers can specify a particular offset point where they can begin to read messages. to GigaSpaces. It enables real-time ingestion of JSON, AVRO, XML, or CSV data from Kafka topics into GigaSpaces spaces, with support for complex transformations, CDC Change Data Capture. A technology that identifies and captures changes made to data in a database, enabling real-time data integration and synchronization between systems. Primarily used for data that is frequently updated, such as user transactions. operations (Insert/Update/Delete), and automatic batching.

Key Features

• Declarative Configuration: Define data pipelines using simple YAML files

• Multiple Data Formats: Support for JSON, AVRO, XML, and CSV

• CDC Support: Automatic handling of Insert, Update, and Delete operations with pluggable CDC solutions (Debezium, HVR, IIDR IBM Infosphere Data Replication. This is a solution to efficiently capture and replicate data, and changes made to the data in real-time from various data sources, including mainframes, and streams them to target systems. For example, used to move data from databases to the In-Memory Data Grid. It is used for Continuous Data Capture (CDC) to keep data synchronized across environments.)

• Complex Data Structures: Full support for nested objects, lists, and their combinations

• Flexible Data Mapping: One Kafka message can map to multiple GigaSpaces types; JSON arrays can be split and distributed

• Data Transformation: Groovy scripting support for preprocessing data before writing to space

• Batch Processing: Collocated task execution with space-side computation

• Scalability The ability of a system to handle increased load by adding resources, such as processing power or storage. Scalability ensures that the system can grow with the demands placed on it.: Horizontally scalable architecture with automatic partition management

• Tiered Storage Automatically assigns data to different categories of storage types based on considerations of cost, performance, availability, and recovery.: Define policies for different space types

• Monitoring: Built-in metrics reporting to InfluxDB and error tracking

• Broadcast Tables: Support for read-only broadcast tables

• Deployment Flexibility: Deploy as a Processing Unit This is the unit of packaging and deployment in the GigaSpaces Data Grid, and is essentially the main GigaSpaces service. The Processing Unit (PU) itself is typically deployed onto the Service Grid. When a Processing Unit is deployed, a Processing Unit instance is the actual runtime entity. (PU This is the unit of packaging and deployment in the GigaSpaces Data Grid, and is essentially the main GigaSpaces service. The Processing Unit (PU) itself is typically deployed onto the Service Grid. When a Processing Unit is deployed, a Processing Unit instance is the actual runtime entity.) or Spring Boot microservice

• Kubernetes An open-source container orchestration system for automating software deployment, scaling, and management of containerized applications. Ready: Helm chart for easy Kubernetes deployment

High-Level Flow

Kafka Topic(s)
				↓
				[Pluggable Connector]
				├─ Reads messages from Kafka
				├─ Parses data format (JSON/AVRO/XML/CSV)
				├─ Applies data transformations (Groovy scripts)
				├─ Determines operation type (Insert/Update/Delete)
				└─ Writes to GigaSpaces space(s)
				↓
				GigaSpaces Space Where GigaSpaces data is stored. It is the logical cache that holds data objects in memory and might also hold them in layered in tiering. Data is hosted from multiple SoRs, consolidated as a unified data model.(s)
				└─ Data available for real-time queries
		

Key Components

1. Kafka Consumer: Reads messages from one or more Kafka topics with configurable offset strategies

2. Data Parser: Parses incoming data in various formats

3. Data Pipeline A series of data processing steps, including extraction, transformation, and loading (ETL), that move data from its source to a destination system. Data pipelines are essential for integrating and managing data flows.: Maps Kafka messages to GigaSpaces space types using JSON Path selectors

4. CDC Engine: Determines operation type (insert/update/delete) based on message content

5. Space Writer: Writes data to GigaSpaces using the Space API

6. Error Handler: Publishes failures to a dedicated error topic for debugging

Prerequisites

Before deploying the Pluggable Connector, ensure you have:

Required Components

1. Apache Kafka: A running Kafka cluster with topics containing your data

   • Minimum version: 2.4.0

   • Verify connectivity and topic availability before starting

2. GigaSpaces: A running GigaSpaces instance

   • Minimum version: 15.0.0 (check your specific release notes)

   • At least one space created where data will be written

3. Java: Java 11 or later installed on your deployment machine

Network Requirements

• Pluggable Connector must be able to reach Kafka brokers on their configured ports (default: 9092)

• Pluggable Connector must be able to reach GigaSpaces lookup service (default: 4174)

• (If using with 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) DI Manager must be able to reach the Pluggable Connector's REST API 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. endpoint

Optional Components

• InfluxDB: For metrics collection and monitoring

• Debezium: For Change Data Capture from databases

• HVR: For Change Data Capture from various sources

Configuration

The Pluggable Connector is configured through two main YAML files:

1. Data Pipeline Configuration (data-pipeline.yml)

This file defines how data from Kafka is mapped to GigaSpaces space types.

Basic Structure

---
				dataFormat: "JSON"              # Data format: JSON, AVRO, XML, or CSV

				cdc:                            # Change Data Capture configuration
				operations:
				insert:                     # Insert operation definition
				defaultOperation: true
				ifExists: "update"
				update:                     # Update operation definition
				conditions:
				- selector: "$.op_type"
				value: 2
				ifNotExists: "insert"
				delete:                     # Delete operation definition
				conditions:
				- selector: "$.op_type"
				value: 0

				spaceTypes:                     # Define space types
				- name: "Product"             # Space type name
				dataSource:
				topic: "products"         # Kafka topic
				properties:
				- name: "id"              # Space property name
				type: "String"          # Property type
				selector: "$.product_id" # JSONPath to extract from message
				spaceid: true           # Mark as space ID
				- name: "name"
				type: "String"
				selector: "$.product_name"
				- name: "price"
				type: "java.math.BigDecimal"
				selector: "$.cost"
				- name: "category"
				type: "String"
				selector: "$.category"
		

Key Configuration Options

Data Format:

• JSON: Recommended for most use cases

• AVRO: For Avro-encoded messages

• XML: For XML documents

• CSV: For comma-separated values

CDC Operations:

• insert: Default operation or when conditions match

  • defaultOperation: true - Apply when no other condition matches

  • ifExists: "update" - If object exists, perform update instead

• update: Applied when conditions match

  • conditions: List of JSONPath conditions

  • ifNotExists: "insert" - If object doesn't exist, insert instead

• delete: Applied when conditions match for removal

Space Types:

• name: The GigaSpaces space type name

• dataSource.topic: Kafka topic to consume from

• properties: Array of mappings from Kafka to space

Property Mapping:

• name: Property name in the space type

• type: Java type (String, Integer, Long, java.math.BigDecimal, etc.)

• selector: JSONPath expression to extract data from message

• spaceid: true: Mark property as unique identifier (use for exactly one property)

• routingkey: true: Mark property for space routing The mechanism that is in charge of routing the objects into and out of the corresponding partitions. The routing is based on a designated attribute inside the objects that are written to the Space, called the Routing Index. (partition key)

Advanced Property Options

properties:
				- name: "nested_object"
				type: "com.example.Address"      # Custom type for nested objects
				selector: "$.address"
  
				- name: "items_list"
				type: "java.util.List"           # List type
				selector: "$.order_items"
  
				- name: "full_message"
				type: "String"
				serializer: "json"               # Serialize entire object as JSON
  
				- name: "dynamic_field"
				type: "String"
				selector: "$.field"
				groovyScript: "value?.toUpperCase()" # Transform value
		

Example: Multi-Type Mapping

spaceTypes:
				- name: "Customer"
				dataSource:
				topic: "customers"
				properties:
				- name: "customerId"
				type: "String"
				selector: "$.id"
				spaceid: true
				- name: "name"
				type: "String"
				selector: "$.name"

				- name: "Order"
				dataSource:
				topic: "orders"
				properties:
				- name: "orderId"
				type: "String"
				selector: "$.order_id"
				spaceid: true
				- name: "customerId"
				type: "String"
				selector: "$.customer_id"
				routingkey: true
				- name: "total"
				type: "java.math.BigDecimal"
				selector: "$.amount"
		

2. Application Configuration (application.yml)

This file configures the connector's runtime behavior, Kafka connection, and space connection.

Basic Structure

---
				# Kafka configuration
				spring.kafka.bootstrap-servers: localhost:9092
				spring.kafka.consumer-group: my-consumer-group

				# GigaSpaces configuration
				space.name: my-space                    # Space to write to
				gs.lookup-locator: localhost:4174       # Lookup locator address
				gs.lookup-groups: "my-group"            # Space group (optional)

				# Connector configuration
				server.port: 6085                       # REST 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. API port

				# Logging
				logging.level:
				root: WARN
				com.gigaspaces.connector: INFO

				---
				# Learning mode configuration (auto-discovery)
				spring.profiles: learning
				spring.kafka.consumer-group: connector-learning
				server.port: 7080

				metadata-provider:
				method: based-on-data
				parser: JSON                          # AUTO, JSON, AVRO, HVR, DEBEZIUM
				topics: [ products, orders ]
				types-mapping:
				use-string-for-unlisted-types: true
				map:
				- source-type: "timestamp"
				dih-type: Long

				---
				# Connector mode configuration (production)
				spring.profiles: connector
				spring.kafka.consumer-group: connector
		

Key Configuration Options

Kafka:

• spring.kafka.bootstrap-servers: Kafka broker addresses (comma-separated)

• spring.kafka.consumer-group: Consumer group ID

• spring.kafka.security.protocol: Security protocol (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL)

• spring.kafka.properties.*: Additional Kafka properties

GigaSpaces:

• space.name: Target space name

• gs.lookup-locator: Lookup service address

• gs.lookup-groups: Space group name

• gs.space-url: Alternative to locator-based lookup

Connector:

• server.port: REST API port for DI Manager communication

• connector.self-url: Self-reported URL (for DI Manager reconciliation)

• di-manager.url: DI Manager base URL (for self-healing)

Learning Mode:

• metadata-provider.method: Discovery method (AUTO, based-on-data)

• metadata-provider.parser: Parser type (AUTO, JSON, AVRO, HVR, DEBEZIUM)

• metadata-provider.topics: Topics to analyze for auto-discovery

Data Mapping

JSONPath Selectors

Data is extracted from Kafka messages using JSONPath expressions in the selector field.

Common JSONPath Patterns

# Root property
				selector: "$.id"                    # Extract message.id

				# Nested property
				selector: "$.user.address.city"    # Extract nested value

				# Array element (first item)
				selector: "$.items[0].name"        # First item's name

				# All array elements (returns array)
				selector: "$.items[*].id"          # All item IDs

				# Properties matching pattern
				selector: "$.order_*"              # Properties starting with "order_"

				# Conditional expression
				selector: "$[?(@.status == 'active')]"  # Items where status is 'active'
		

Example Data Mapping

Input Kafka Message (JSON):

{
				"event_id": "evt-123",
				"timestamp": 1234567890,
				"customer": {
				"id": "cust-456",
				"name": "John Doe",
				"email": "john@example.com"
				},
				"order": {
				"id": "ord-789",
				"items": [
				{"product_id": "prod-1", "quantity": 2},
				{"product_id": "prod-2", "quantity": 1}
				],
				"total": 99.99
				}
				}
		

Configuration:

spaceTypes:
				- name: "Customer"
				dataSource:
				topic: "orders"
				properties:
				- name: "customerId"
				type: "String"
				selector: "$.customer.id"
				spaceid: true
				- name: "name"
				type: "String"
				selector: "$.customer.name"
				- name: "email"
				type: "String"
				selector: "$.customer.email"
		

Result in space:

Customer {
				customerId: "cust-456"
				name: "John Doe"
				email: "john@example.com"
				}
		

Type Conversion

The connector automatically converts JSON values to the specified Java types:

YAML Type	Java Class	Example
String	java.lang.String	"hello"
Integer	java.lang.Integer	42
Long	java.lang.Long	9223372036854775807
Double	java.lang.Double	3.14
BigDecimal	java.math.BigDecimal	"99.99"
Boolean	java.lang.Boolean	true
Date	java.util.Date	1234567890 (timestamp in ms)
Timestamp	java.sql.Timestamp	1234567890

Handling Null Values

properties:
				- name: "optional_field"
				type: "String"
				selector: "$.maybe_missing"
				nullable: true              # Allow null values
				defaultValue: "N/A"         # Use when field is missing
		

Nested Objects

properties:
				- name: "address"
				type: "com.example.Address"  # Custom class
				selector: "$.customer.address"
  
				- name: "metadata"
				type: "java.util.Map"        # Store as Map
				selector: "$.extra"
		

Arrays and Lists

properties:
				- name: "tags"
				type: "java.util.List"
				selector: "$.tags"
  
				- name: "items"
				type: "java.util.List"
				selector: "$.order_items"
		

CDC Operations

Change Data Capture (CDC) determines whether a message results in an Insert, Update, or Delete operation in the space.

CDC Configuration Structure

cdc:
				operations:
				insert:
				defaultOperation: bool     # Is this the default?
				ifExists: "update|delete"  # What to do if object exists
				conditions:
				- selector: JSONPath     # Condition 1
				value: anyValue
				- selector: JSONPath     # Condition 2
				value: anyValue
				ifNotExists: "insert"      # What to do if doesn't exist
  
				update:
				conditions: [...]
				ifNotExists: "insert"
  
				delete:
				conditions: [...]
		

How CDC Works

1. Evaluate Conditions: Check each operation's conditions against the message

2. Match Operation: First matching operation determines the action

3. Apply Fallback: If no condition matches, use defaultOperation: true

4. Handle Existence: Consider ifExists and ifNotExists modifiers

Example: Debezium CDC

cdc:
				operations:
				insert:
				defaultOperation: false
				conditions:
				- selector: "$.op"
				value: "c"             # Debezium create
				ifExists: "update"
  
				update:
				conditions:
				- selector: "$.op"
				value: "u"             # Debezium update
				ifNotExists: "insert"
  
				delete:
				conditions:
				- selector: "$.op"
				value: "d"             # Debezium delete
		

Example: HVR CDC

cdc:
				operations:
				insert:
				defaultOperation: true     # Default if no condition matches
				ifExists: "update"
  
				update:
				conditions:
				- selector: "$.payload.hvr_operation_type"
				value: 2               # HVR update code
				ifNotExists: "insert"
  
				delete:
				conditions:
				- selector: "$.payload.hvr_operation_type"
				value: 0               # HVR delete code
		

Example: Simple Timestamp-Based CDC

cdc:
				operations:
				insert:
				defaultOperation: true
				conditions:
				- selector: "$.event_type"
				value: "created"
  
				update:
				conditions:
				- selector: "$.event_type"
				value: "modified"
  
				delete:
				conditions:
				- selector: "$.event_type"
				value: "deleted"
		

Advanced Features

Groovy Script Transformations

Transform values before writing to space using Groovy scripts:

properties:
				- name: "full_name"
				type: "String"
				groovyScript: "firstName + ' ' + lastName"
  
				- name: "amount_cents"
				type: "Long"
				selector: "$.amount"
				groovyScript: "value * 100"  # Convert dollars to cents
  
				- name: "status_upper"
				type: "String"
				selector: "$.status"
				groovyScript: "value?.toUpperCase() ?: 'UNKNOWN'"
		

Available variables in scripts:

• value: The extracted value from the message

• message: The entire Kafka message

• payload: The parsed payload object

• Any property previously processed in the same message

Message Serialization

Store entire message parts as JSON strings:

properties:
				- name: "raw_data"
				type: "String"
				serializer: "json"           # Serialize as JSON string
				selector: "$.extra_fields"
		

Broadcast Tables

Define read-only broadcast tables replicated across all partitions:

spaceTypes:
				- name: "LookupTable"
				broadcast: true              # Mark as broadcast
				dataSource:
				topic: "lookup"
				properties:
				- name: "code"
				type: "String"
				selector: "$.code"
				spaceid: true
		

Tiered Storage Policies

Define storage policies for space types:

spaceTypes:
				- name: "HighVolumeData"
				storage-policy:
				type: "tiered"
				memoryThreshold: 80        # % of heap
				spillPolicy: "disk"        # Spill to disk
				dataSource:
				topic: "events"
				properties: [...]
		

Batch Processing

Configure batch parameters for optimized ingestion:

connector:
				batch:
				size: 1000                   # Process 1000 messages per batch
				timeoutMs: 5000              # Or timeout after 5 seconds
				parallelism: 4               # Process 4 batches concurrently
		

Error Handling

Configure error handling and reporting:

connector:
				error-handling:
				strategy: "dlq"              # dead-letter-queue or skip
				dlq-topic: "connector-errors" # Topic for errors
				max-retries: 3
				retry-backoff-ms: 1000
		

Errors are published with the structure:

{
				"pipelineId": "pipeline-id",
				"timestamp": 1234567890,
				"originalMessage": {...},
				"error": "error message",
				"stackTrace": "..."
				}
		

Installation

Option 1: Docker Deployment (Recommended for Development)

# Clone the repository
				git clone https://github.com/giga-di/di-kafka-connector.git
				cd di-kafka-connector

				# Build the project
				mvn clean install

				# Navigate to an example
				cd examples/cdc-debezium-mssql-standalone

				# Start the environment (includes Kafka, Database, GigaSpaces)
				./1-launch-environment-and-wait.sh

				# Run the connector
				./6-start-connector.sh
		

Option 2: Kubernetes with Helm (Production)

1. Add the Helm repository:

   helm repo add pluggable-connector <your-repo-url>
   						helm repo update
   				

2. Create a values file (values.yaml):

   image:
   						repository: your-registry/pluggable-connector
   						tag: "17.2.2"
   
   						kafka:
   						brokers: "kafka-broker:9092"
   
   						space:
   						lookupLocator: "gigaspaces-service:4174"
   						space: "my-space"
   
   						connector:
   						port: 6085
   						selfUrl: "http://pluggable-connector:6085"
   
   						diManager:
   						url: "http://di-manager:8080"
   				

3. Install the chart:

   helm install pluggable-connector ./helm-chart/pluggable-connector \
   						-f values.yaml \
   						-n di-kafka \
   						--create-namespace
   				

Option 3: Manual Spring Boot Deployment

1. Build the Spring Boot JAR:

   mvn -DskipTests=true clean package
   						cd spring-boot
   						mvn spring-boot:run
   				

2. Configure via environment variables:

   export SPRING_KAFKA_BOOTSTRAP_SERVERS=localhost:9092
   						export SPACE_NAME=my-space
   						export GS_LOOKUP_LOCATOR=localhost:4174
   						export SERVER_PORT=6085
   
   						java -jar spring-boot/target/pluggable-connector-*.jar
   				

Running the Connector

Starting the Connector

Using Docker:

docker run -d \
				--name pluggable-connector \
				-e SPRING_KAFKA_BOOTSTRAP_SERVERS=kafka:9092 \
				-e SPACE_NAME=my-space \
				-e GS_LOOKUP_LOCATOR=gigaspaces:4174 \
				-p 6085:6085 \
				your-registry/pluggable-connector:17.2.2
		

Using Kubernetes:

kubectl apply -f pluggable-connector-deployment.yaml
		

Manually:

java -jar pluggable-connector.jar \
				--spring.kafka.bootstrap-servers=kafka:9092 \
				--space.name=my-space \
				--gs.lookup-locator=gigaspaces:4174
		

Verifying the Connector is Running

1. Check the health endpoint:

   curl http://localhost:6085/v1/info
   				

   Expected response:

   {
   						"componentName": "Pluggable-Connector",
   						"version": "17.2.2",
   						"status": "RUNNING",
   						"host": "connector-host",
   						"pid": 12345
   						}
   				

2. Check the logs:

   kubectl logs -f deployment/pluggable-connector
   				

3. Verify Kafka connectivity:

   • Connector should connect to Kafka without errors in logs

   • Look for: Connected to Kafka broker

4. Verify GigaSpaces connectivity:

   • Connector should connect to the space

   • Look for: Connected to space: my-space

Stopping the Connector

# Docker
				docker stop pluggable-connector

				# Kubernetes
				kubectl delete deployment pluggable-connector

				# Manual
				pkill -f "pluggable-connector"
		

Connector URL

The Pluggable Connector has a REST controller. It is exposed as a service and listens by default on port 6085.

http://<helm release name>-pluggable-connector:6085
		

Relationship to DI Manager

The Pluggable Connector can be controlled via DI Manager, which orchestrates (create, start, stop, delete, and list) Kafka-to-Space pipelines, forwarding start/stop commands to the underlying connector.

Starting the Pluggable Connector

Run from inside the cluster (via kubectl exec into the pc pod):

PC_POD=$(kubectl get pod -n <namespace> -l app.kubernetes.io/name=pluggable-connector -o jsonpath='{.items[0].metadata.name}')
				PIPELINE_ID=$(python3 -c "import uuid;print(uuid.uuid4())")

				kubectl exec -n <namespace> "$PC_POD" -- \
				curl -s -X POST "http://localhost:6085/v1/control/${PIPELINE_ID}/start?offsetStrategy=EARLIEST&topic=customers"
		

Or from your local machine (port-forward first):

kubectl port-forward -n <namespace> svc/<helm release name>-pluggable-connector 6085:6085 &
				PIPELINE_ID=$(python3 -c "import uuid;print(uuid.uuid4())")

				curl -X POST "http://localhost:6085/v1/control/${PIPELINE_ID}/start?offsetStrategy=EARLIEST&topic=customers"
		

For direct/manual operation, see the note above about PIPELINE_ID when DI Manager owns the pipeline.

Control API (Start / Stop / Status)

# Start a pipeline
				curl -X POST "http://<helm release name>-pluggable-connector:6085/v1/control/${PIPELINE_ID}/start?offsetStrategy=EARLIEST&topic=customers"

				# Stop a pipeline
				curl -X POST "http://<helm release name>-pluggable-connector:6085/v1/control/${PIPELINE_ID}/stop"

				# Check status
				curl "http://<helm release name>-pluggable-connector:6085/v1/control/${PIPELINE_ID}/status"
		

The topic parameter tells the connector which pipeline UUID owns which Kafka topic, so InfluxDB metrics are tagged with the correct pipelineId for 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. queries.