DEV Community: maghsood esmaeili

How to Sync Data from an Oracle Table to Elasticsearch using Kafka Connect

maghsood esmaeili — Tue, 16 Dec 2025 06:31:22 +0000

In this document, I’ll walk you through the challenge I faced when fetching data from an Oracle database, streaming it into Kafka, and finally consuming and writing that data into Elasticsearch. My goal is that this guide will help other teams build a reliable data pipeline using similar components.
Before we begin, you’ll need a working Kafka setup. In my case, I prepared a Kafka cluster and verified that all the pods were running correctly. This ensures the environment is ready before configuring Kafka Connect and the connectors for Oracle and Elasticsearch.
The general flow we’ll cover is:

Prerequisite: Installing and running a Kafka cluster using the Strimzi operator.
Configure Kafka Connect: Create a Kafka Connect instance.
Create a custom Dockerfile: Create a custom image and push it to the local registry.
Configure JDBC Connector: Configure the JDBC connector in the Kafka cluster.
Configure ElasticSearch Connector: Add sink configuration to insert and create an index in ElasticSearch.
Configure ingest pipeline(optional): convert Oracle timestamp field to Elastic
search timestamp field.

By the end of this guide, you should have a running pipeline that automatically streams changes from Oracle to Elasticsearch with minimal manual intervention.
Before starting, ensure that Kafka is installed and running in your cluster. In my setup, I deployed a Kafka cluster and confirmed that all pods are up and running:

Afterward, Kafka Connect must be installed using the Strimzi operator. This requires creating a Kafka Connect Custom Resource Definition (CRD) instance on the cluster. The following configuration provides a sample cluster file:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  annotations:
    strimzi.io/use-connector-resources: 'true'
  name: my-connect-cluster
  namespace: kafka-infra
spec:
  bootstrapServers: 'my-cluster-kafka-bootstrap:9092'
  config:
    config.storage.replication.factor: -1
    config.storage.topic: connect-cluster-configs
    group.id: connect-cluster
    offset.storage.replication.factor: -1
    offset.storage.topic: connect-cluster-offsets
    status.storage.replication.factor: -1
    status.storage.topic: connect-cluster-status
  image: 'private-registery.com/strimzi/kafka-custom:v2'
  replicas: 1
  version: 4.0.0

The next step, and one of the most critical components of the data pipeline, is the installation of the ElasticsearchSink and ElasticsearchSink plugins within the Kafka Connect pod. By default, Kafka Connect does not include these plugins, which means a custom image must be built.
This process involves creating a Docker image that packages the required connector plugins and then updating the Kafka Connect custom resource to reference this new image.

Doing so ensures that Kafka Connect can interact with both Elasticsearch and relational databases, enabling seamless data integration across the pipeline. The following sample Dockerfile demonstrates how to add the ElasticsearchSink and JDBCConnector plugins to the image:

FROM quay.io/strimzi/kafka:0.46.0-kafka-4.0.0

USER root
RUN curl -o /tmp/confluentinc-kafka-connect-elasticsearch-15.0.1.zip https://hub-downloads.confluent.io/api/plugins/confluentinc/kafka-connect-elasticsearch/versions/15.0.1/confluentinc-kafka-connect-elasticsearch-15.0.1.zip

RUN curl -o /tmp/confluentinc-kafka-connect-jdbc-10.8.4.zip https://hub-downloads.confluent.io/api/plugins/confluentinc/kafka-connect-jdbc/versions/10.8.4/confluentinc-kafka-connect-jdbc-10.8.4.zip

RUN unzip /tmp/confluentinc-kafka-connect-jdbc-10.8.4.zip -d /opt/kafka/plugins/ && \
    rm /tmp/confluentinc-kafka-connect-jdbc-10.8.4.zip


RUN unzip /tmp/confluentinc-kafka-connect-elasticsearch-15.0.1.zip -d /opt/kafka/plugins/ && \
    rm /tmp/confluentinc-kafka-connect-elasticsearch-15.0.1.zip

USER 1001

Finally, we need to add a Kafka connector to capture data from Oracle and publish it to Kafka. The configuration below illustrates how to define the JDBC connector configuration file:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  labels:
    strimzi.io/cluster: my-connect-cluster
  name: my-source-connector-jdbc-testdb2-v0
  namespace: kafka-infra
spec:
  class: io.confluent.connect.jdbc.JdbcSourceConnector
  config:
    poll.interval.ms: 3600000
    transforms.extractInt.field: ORACLE_TIME_FIELD
    mode: timestamp
    query: select * from schema_name.table_name
    timestamp.column.name: ORACLE_TIME_FIELD
    connection.password: *****
    topic.prefix: oracle-testdb2-audit-table
    connection.user: thinuser
    connection.url: 'jdbc:oracle:driver-name:@hostname:port/db-name' 
  tasksMax: 2

We can now verify in Kafka that the Oracle table topic defined in the JDBC connector configuration has been created successfully and is receiving messages :
topic name(by running):

bin/kafka-topics.sh --list --bootstrap-server localhost:9092
output:
oracle-testdb2-audit-table
messages(by running):

bin/kafka-console-consumer.sh -–topic oracle-testdb2-audit-table --bootstrap-server localhost:9092 --from-beginning --max-messages 1

Later, I will provide a Kafka consumer implemented in Go.
The next section of the documentation will cover how to synchronize data from Kafka to Elasticsearch.
First, we need to install a Kafka connector to fetch data from Kafka and synchronize it with Elasticsearch. The configuration for the Kafka connector is shown below:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  labels:
    strimzi.io/cluster: my-connect-cluster
  name: my-source-connector-oracle-v1
  namespace: kafka-infra
spec:
  class: io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
  config:
    key.ignore: true
    value.converter: org.apache.kafka.connect.json.JsonConverter
    connection.username: elastic
    value.converter.schemas.enable: false
    name: elasticsearch-sink-connector
    connection.password: ******
    key.converter: org.apache.kafka.connect.storage.StringConverter
    drop.invalid.message: true
    behavior.on.malformed.documents: ignore
    retry.backoff.ms: 1000
    behavior.on.null.values: ignore
    max.retries: 5
    topics.regex: oracle-test.*
    type.name: _doc
    connection.url: 'http://elasticsearch-hostname:9200'
    schema.ignore: true
  tasksMax: 2

After installing the connector, the Elasticsearch index will be created automatically. The image below illustrates the newly created index in Elasticsearch, with the document count field indicating the amount of data inserted into the cluster.

The image shows the oracle-testdb2-audit-table index created.

Optional: After completing the steps outlined above, we encountered an additional challenge related to the timestamp field. The timestamp column defined in the Oracle database was of type TIMESTAMP, but after processing through the data pipeline (from Oracle to Elasticsearch), Elasticsearch interpreted this field as a Long type. The following steps should be performed to resolve this issue.

Create a new Elasticsearch index. The image below demonstrates how to define a custom Elasticsearch index:

Add an ingest pipeline in Elasticsearch to convert a Long-type field into a Date-type field. The image below illustrates how to define the ingest pipeline:

Attach the ingest pipeline to the Elasticsearch index.

summery

This document describes how to build a data pipeline that synchronizes data from an Oracle database to Elasticsearch using Kafka and Kafka Connect. It begins with preparing a Kafka cluster deployed via the Strimzi operator and creating a Kafka Connect instance.

The pipeline first uses a JDBC Source Connector to read data from an Oracle table and publish it to Kafka topics. The setup is verified by checking topic creation and consuming sample messages from Kafka. Next, an Elasticsearch Sink Connector is configured to consume data from Kafka and automatically create and populate an Elasticsearch index.

Kubernetes validating admission policy and admission binding

maghsood esmaeili — Tue, 04 Nov 2025 07:41:30 +0000

ValidatingAdmissionPolicy is a new Kubernetes plugin designed to manage access to Kubernetes resources. It validates which users, groups, or service accounts are allowed to perform specific actions — such as CREATE, DELETE, UPDATE, or CONNECT — on various Kubernetes resources.

It helps reduce malicious activities within the Kubernetes cluster and enhances overall security. Acting as a gatekeeper in front of Kubernetes resources, it ensures that only authenticated and authorized requests are allowed to perform actions on the cluster.

Validating admission policies uses the Common Expression Language (CEL) to declare the validation rules of a policy.

When a request is sent from the API server to apply or modify Kubernetes resources, the Validating Admission Webhooks intercept it to prevent unauthenticated or invalid requests. The image below illustrates the detailed flow of the Kubernetes API server.

Let’s see why we might need to use a Validating Admission Policy in our Kubernetes cluster.

Use case: We want only the DevOps group to have permission to create, update, or delete on ArgoCD Custom Resources (Argocds), while preventing other groups in the cluster from performing these actions.

One practical way to achieve this is by creating a ValidatingAdmissionPolicy and ValidatingAdmissionPolicyBinding resources at the cluster level.

This resource is introduced at the cluster-wide level.

Below is a sample Validating Admission Policy:

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingAdmissionPolicy
metadata:
  name: check-argocds-operation
  namespace: kube-system
spec:
  matchConstraints:
    resourceRules:
      - apiGroups:
          - "argoproj.io"
        apiVersions:
          - "v1alpha1"
        operations: 
          - "CREATE"
          - "UPDATE"
          - "DELETE"
        resources:
          - "argocds"
  failurePolicy: Fail
  variables:
  - name: requestedUsername
    expression: 'request.userInfo.username'


  validations:
    - expression: '("devops-group" in request.userInfo.groups) || ( request.userInfo.username == "system:serviceaccount:openshift-gitops-operator:openshift-gitops-operator-controller-manager") '
      messageExpression: >-
        variables.requestedUsername

Explanation of the YAML Sample

**Here’s a brief explanation of the YAML manifest I created:

resourceRules:
This is one of the most important parts of the manifest. You must define the apiGroups and apiVersions for your target resource. (For more details about Kubernetes components, refer to the official Kubernetes documentation.)

Operations:
Specifies which actions (e.g., CREATE, UPDATE, DELETE) should trigger validation.

Resources:
Defines which Kubernetes resources the policy applies to. In this example, the resource is argocds, which we want to validate.

failurePolicy (optional):
Determines how the webhook behaves if it fails. Options include:

Fail — Reject the request.
Ignore — Allow the request to proceed, skipping webhook validation.

variables (optional):
You can define variables to use in the messageExpression.This helps display clearer messages for users who don’t have permission to perform certain actions. It’s also useful for debugging the contents of your request object.

validations:
Contains the list of validation expressions. In this example, the policy checks the requester’s username and group. The expression used here was tested on OpenShift.

The message expression defined earlier is used to display a clear message to users who lack sufficient permissions to act.

Finally, to apply the policy to the cluster, we need to create a ValidatingAdmissionPolicyBinding object. The YAML example below demonstrates how to use it.

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingAdmissionPolicyBinding
metadata:
  name: argocds-operation-validating-binding
spec:
  policyName: check-argocds-operation
  validationActions:
    - "Deny" # Warn, Audit

validationActions: If the validation expression (defined in the ValidatingAdmissionPolicy) is false, it means the user does not have sufficient permissions to apply the specified action. In this example, the action denies the user from performing operations on the cluster.

Summary
ValidatingAdmissionPolicies in Kubernetes controls access to cluster resources by determining which users, groups, or service accounts can perform actions. They act as a gatekeeper, reducing malicious activity and ensuring that only authenticated and authorized requests are allowed. We discussed the ValidatingAdmissionPolicy plugin and explored an example demonstrating its usage in the Kubernetes cluster. To enforce the policy across the cluster, it is necessary to create a ValidatingAdmissionPolicyBinding resource.

From Monoliths to Microservices: The Role of Service Mesh in Modern Applications

maghsood esmaeili — Tue, 28 Oct 2025 07:23:14 +0000

A service mesh is a dedicated infrastructure layer that manages communication between microservices within an application. It provides features such as traffic routing, security, observability, and resilience, allowing developers to offload these cross-cutting concerns from individual services. By doing so, a service mesh simplifies service-to-service interactions and improves the reliability and manageability of distributed systems.
Let’s start by understanding what monoliths and microservices are, how they differ, and why a service mesh is important when running applications in production.

Monoliths Microservices

A monolithic application typically deploys all functionalities together within a unified codebase, with minimal separation between components. This tight coupling often leads to issues such as a single database acting as a performance bottleneck.
In a monolithic service, each application depends on a specific version of another application.

Even minor updates require a full redeployment, making scalability and independent upgrades challenging.
So suppose the product team wants to test a new feature in production for a specific rate of requests. It's not possible with this architecture.

In larger enterprise applications with hundreds of modules maintained by numerous developers, loosely defined architectural rules can quickly transform the system into what is often referred to as a "big ball of mud"—an unmanageable, complex codebase.

Transitioning to Microservices

This transformation is complex and requires cultural, technical, and organizational shifts toward cloud-native practices.
In a microservices architecture, each module becomes an independent application and can be implemented with different programming languages.

This microservices approach offers several benefits:

Independent Scaling
Faster Releases
Technological Flexibility
Enhanced Resilience
Manageability

However, moving to microservices also introduces challenges. In the monolith, functionalities such as networking, authentication, authorization, data transfer, logging, monitoring, and tracing were centrally managed. With microservices, these cross-cutting concerns are duplicated across independent teams, leading to increased complexity in managing certificates, monitoring agents, traffic rules, timeouts, and service discovery.

Service Mesh

Now, let’s go back to the idea of the service mesh and take a closer look at it, as I mentioned before. In traditional microservice architectures, each service was tasked with handling its own routing, security, and observability functions. With a service mesh, these tasks are offloaded to sidecar proxies deployed alongside every microservice. The network communication between services is managed by these proxies, which also form the data plane.
The proxies communicate with a central server-side component known as the Control Plane. The Control Plane oversees and directs all traffic entering and leaving the services, ensuring a larger, cohesive system.

Key Benefits

Dynamic configuration of service interactions without direct code changes.
Enhanced security through mutual TLS, protecting communications between services.
Comprehensive observability, enabling real-time monitoring, performance assessments, and bottleneck detection.
By abstracting the networking logic into a separate infrastructure, a service mesh allows you to dynamically configure and manage the interactions between services. This means enhanced security, improved scalability, and more efficient traffic management, leading to robust and resilient application performance.

Example Implementations

Istio: Provides advanced traffic management, observability (metrics, logs, traces), and security (mTLS, policy enforcement).
Linkerd: Lightweight and performance-focused mesh designed for simplicity and speed.
Consul: Offers service discovery, configuration, and segmentation capabilities in addition to service mesh features.
AWS App Mesh: A managed service mesh that integrates with AWS services like ECS and EKS for consistent communication management.

conclusion

The evolution from monolithic to microservices architectures has greatly enhanced scalability, flexibility, and resilience, but it has also introduced new challenges in managing communication, security, and observability across distributed services. A service mesh effectively addresses these challenges by providing a dedicated layer that centralizes control over service-to-service interactions. Through features like dynamic traffic routing, mutual TLS, and real-time monitoring, it simplifies operations and strengthens the reliability of modern cloud-native systems. Technologies such as Istio, Linkerd, Consul, and AWS App Mesh exemplify how service meshes are shaping the future of microservices management.