diff --git a/.github/workflows/ci-cd.yml b/.github/workflows/ci-cd.yml
new file mode 100644
index 0000000..195b0e2
--- /dev/null
+++ b/.github/workflows/ci-cd.yml
@@ -0,0 +1,169 @@
+name: CI/CD Pipeline
+
+on:
+  push:
+    branches: [ main, develop ]
+    tags: [ 'v*' ]
+  pull_request:
+    branches: [ main ]
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up JDK 17
+      uses: actions/setup-java@v4
+      with:
+        java-version: '17'
+        distribution: 'temurin'
+
+    - name: Cache Gradle packages
+      uses: actions/cache@v4
+      with:
+        path: |
+          ~/.gradle/caches
+          ~/.gradle/wrapper
+        key: ${{ runner.os }}-gradle-${{ hashFiles('**/*.gradle*', '**/gradle-wrapper.properties') }}
+        restore-keys: |
+          ${{ runner.os }}-gradle-
+
+    - name: Make gradlew executable
+      run: chmod +x ./gradlew
+
+    - name: Run tests
+      run: ./gradlew test jacocoTestReport
+
+    - name: Check test coverage
+      run: ./gradlew jacocoTestCoverageVerification
+
+    - name: Upload test results
+      uses: actions/upload-artifact@v4
+      if: always()
+      with:
+        name: test-results
+        path: |
+          build/reports/tests/
+          build/reports/jacoco/
+
+  build-and-push:
+    needs: test
+    runs-on: ubuntu-latest
+    outputs:
+      image-tag: ${{ steps.meta.outputs.tags }}
+      image-digest: ${{ steps.build.outputs.digest }}
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up Docker Buildx
+      uses: docker/setup-buildx-action@v3
+
+    - name: Log in to Container Registry
+      uses: docker/login-action@v3
+      with:
+        registry: ${{ env.REGISTRY }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
+
+    - name: Extract metadata
+      id: meta
+      uses: docker/metadata-action@v5
+      with:
+        images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+        tags: |
+          type=ref,event=branch
+          type=ref,event=pr
+          type=semver,pattern={{version}}
+          type=semver,pattern={{major}}.{{minor}}
+          type=semver,pattern=1.0.{{patch}}
+          type=raw,value=latest,enable={{is_default_branch}}
+
+    - name: Build and push Docker image
+      id: build
+      uses: docker/build-push-action@v5
+      with:
+        context: .
+        platforms: linux/amd64,linux/arm64
+        push: true
+        tags: ${{ steps.meta.outputs.tags }}
+        labels: ${{ steps.meta.outputs.labels }}
+        cache-from: type=gha
+        cache-to: type=gha,mode=max
+
+  deploy:
+    needs: build-and-push
+    runs-on: ubuntu-latest
+    if: github.ref == 'refs/heads/main' || startsWith(github.ref, 'refs/tags/v')
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up Helm
+      uses: azure/setup-helm@v4
+      with:
+        version: '3.13.0'
+
+    - name: Set up kubectl
+      uses: azure/setup-kubectl@v4
+      with:
+        version: '1.28.0'
+
+    - name: Configure kubectl for IKS
+      run: |
+        echo "${{ secrets.KUBECONFIG }}" | base64 -d > kubeconfig
+        export KUBECONFIG=kubeconfig
+        kubectl config current-context
+
+    - name: Deploy to Kubernetes
+      env:
+        KUBECONFIG: kubeconfig
+        SENDGRID_API_KEY: ${{ secrets.SENDGRID_API_KEY }}
+        KAFKA_BOOTSTRAP_SERVERS: ${{ secrets.KAFKA_BOOTSTRAP_SERVERS }}
+      run: |
+        # Extract version from tag or use latest
+        if [[ $GITHUB_REF == refs/tags/v* ]]; then
+          VERSION=${GITHUB_REF#refs/tags/v}
+        else
+          VERSION="latest"
+        fi
+        
+        # Deploy using Helm
+        helm upgrade --install kafka-sendgrid ./helm/kafka-sendgrid \
+          --namespace kafka-sendgrid \
+          --create-namespace \
+          --set image.tag=${VERSION} \
+          --set secrets.sendgridApiKey=${SENDGRID_API_KEY} \
+          --set config.kafka.bootstrapServers=${KAFKA_BOOTSTRAP_SERVERS} \
+          --wait \
+          --timeout=300s
+
+    - name: Verify deployment
+      env:
+        KUBECONFIG: kubeconfig
+      run: |
+        kubectl get pods -n kafka-sendgrid
+        kubectl rollout status deployment/kafka-sendgrid -n kafka-sendgrid
+
+  security-scan:
+    needs: build-and-push
+    runs-on: ubuntu-latest
+    steps:
+    - name: Run Trivy vulnerability scanner
+      uses: aquasecurity/trivy-action@master
+      with:
+        image-ref: ${{ needs.build-and-push.outputs.image-tag }}
+        format: 'sarif'
+        output: 'trivy-results.sarif'
+
+    - name: Upload Trivy scan results to GitHub Security tab
+      uses: github/codeql-action/upload-sarif@v3
+      if: always()
+      with:
+        sarif_file: 'trivy-results.sarif'
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..74b76b0
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,46 @@
+# Gradle
+.gradle/
+build/
+!gradle/wrapper/gradle-wrapper.jar
+
+# Java
+*.class
+*.jar
+*.war
+*.nar
+*.ear
+*.zip
+*.tar.gz
+*.rar
+
+# IDE
+.idea/
+*.iws
+*.iml
+*.ipr
+out/
+.vscode/
+
+# OS
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db
+
+# Logs
+*.log
+
+# Test Reports
+/reports/
+
+# Temporary files
+*.tmp
+*.swp
+*.bak
+
+# Environment variables
+.env
+.env.local
\ No newline at end of file
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..8a65b49
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,52 @@
+# Build stage
+FROM gradle:8.11.1-jdk17-alpine AS build
+
+# Set working directory
+WORKDIR /app
+
+# Copy gradle files
+COPY build.gradle .
+COPY gradle ./gradle
+COPY gradlew .
+COPY settings.gradle* ./
+
+# Copy source code
+COPY src ./src
+
+# Build the application
+RUN ./gradlew build --no-daemon -x test
+
+# Runtime stage - using Java 17 Alpine 3.21
+FROM eclipse-temurin:17-jre-alpine
+
+# Install dumb-init for proper signal handling
+RUN apk add --no-cache dumb-init
+
+# Create application user
+RUN addgroup -g 1001 -S appgroup && \
+    adduser -u 1001 -S appuser -G appgroup
+
+# Set working directory
+WORKDIR /app
+
+# Copy the built JAR from build stage
+COPY --from=build /app/build/libs/kafka-sendgrid-*.jar app.jar
+
+# Change ownership of the app directory
+RUN chown -R appuser:appgroup /app
+
+# Switch to non-root user
+USER appuser
+
+# Expose the port
+EXPOSE 8080
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+  CMD wget --no-verbose --tries=1 --spider http://localhost:8080/actuator/health || exit 1
+
+# Use dumb-init to properly handle signals
+ENTRYPOINT ["dumb-init", "--"]
+
+# Run the application
+CMD ["java", "-Djava.security.egd=file:/dev/./urandom", "-jar", "app.jar"]
\ No newline at end of file
diff --git a/README.md b/README.md
index 8b4b25e..f8a6c1f 100644
--- a/README.md
+++ b/README.md
@@ -7,13 +7,16 @@ A Spring Boot 3.5.4 application that consumes messages from a Kafka topic and se
 * [Description](#description)
 * [Prerequisites](#prerequisites)
 * [Getting Started](#getting-started)
-
   * [Clone the Repository](#clone-the-repository)
   * [Configuration](#configuration)
   * [Build and Run](#build-and-run)
 * [Application Properties](#application-properties)
 * [Dependencies](#dependencies)
 * [Usage](#usage)
+* [Testing](#testing)
+* [Docker](#docker)
+* [Kubernetes Deployment](#kubernetes-deployment)
+* [CI/CD](#cicd)
 * [Logging](#logging)
 * [Contributing](#contributing)
 * [License](#license)
@@ -25,7 +28,7 @@ The **kafka-sendgrid** service is a lightweight Spring Boot application (version
 ## Prerequisites
 
 * Java 17 or higher
-* Maven 3.8+ or Gradle 7+
+* Gradle 8+ (included via wrapper)
 * Access to a running Kafka cluster
 * A valid SendGrid API key
 
@@ -34,110 +37,186 @@ The **kafka-sendgrid** service is a lightweight Spring Boot application (version
 ### Clone the Repository
 
 ```bash
-git clone https://github.com/your-org/kafka-sendgrid.git
+git clone https://github.com/cspb1913/kafka-sendgrid.git
 cd kafka-sendgrid
 ```
 
 ### Configuration
 
-Copy the sample configuration file and update with your values:
+The application uses environment variables for configuration. Create your own configuration or use the default values:
 
-```bash
-cp src/main/resources/application.yml.example src/main/resources/application.yml
-```
+#### Environment Variables
 
-Edit `src/main/resources/application.yml`:
-
-```yaml
-spring:
-  kafka:
-    bootstrap-servers: localhost:9092
-    consumer:
-      group-id: kafka-sendgrid-group
-      auto-offset-reset: earliest
-    topic:
-      name: sendgrid-topic
-
-sendgrid:
-  api-key: YOUR_SENDGRID_API_KEY
-  from-email: no-reply@yourdomain.com
-```
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `KAFKA_BOOTSTRAP_SERVERS` | Kafka broker addresses | `localhost:9092` |
+| `KAFKA_CONSUMER_GROUP_ID` | Consumer group ID | `kafka-sendgrid-group` |
+| `KAFKA_AUTO_OFFSET_RESET` | Offset reset strategy | `earliest` |
+| `KAFKA_TOPIC_NAME` | Kafka topic to consume from | `sendgrid-topic` |
+| `SENDGRID_API_KEY` | SendGrid API key | `your-sendgrid-api-key` |
+| `SENDGRID_FROM_EMAIL` | Default sender email | `no-reply@yourdomain.com` |
 
 ### Build and Run
 
-Build the project with Maven:
+Build the project:
 
 ```bash
-mvn clean package -DskipTests
+./gradlew build
 ```
 
 Run the application:
 
 ```bash
-java -jar target/kafka-sendgrid-0.0.1-SNAPSHOT.jar
+./gradlew bootRun
 ```
 
-Or using Gradle:
+Or run the JAR directly:
 
 ```bash
-gradle bootRun
+java -jar build/libs/kafka-sendgrid-1.0.0.jar
 ```
 
 ## Application Properties
 
-| Property                                  | Description                                |
-| ----------------------------------------- | ------------------------------------------ |
-| `spring.kafka.bootstrap-servers`          | Comma-separated list of Kafka brokers      |
-| `spring.kafka.consumer.group-id`          | Consumer group ID                          |
-| `spring.kafka.consumer.auto-offset-reset` | Offset reset strategy (e.g., `earliest`)   |
-| `spring.kafka.topic.name`                 | Name of the Kafka topic to consume         |
-| `sendgrid.api-key`                        | SendGrid API key                           |
-| `sendgrid.from-email`                     | Sender email address for outgoing messages |
+All configuration is handled via environment variables. See the `application.yml` file for the complete configuration structure.
 
 ## Dependencies
 
-Key dependencies declared in `pom.xml` or `build.gradle`:
+Key dependencies:
 
-* `org.springframework.boot:spring-boot-starter`
-* `org.springframework.boot:spring-boot-starter-kafka`
-* `com.sendgrid:sendgrid-java`
-* `org.springframework.boot:spring-boot-starter-logging`
+* `org.springframework.boot:spring-boot-starter` - Core Spring Boot
+* `org.springframework.boot:spring-boot-starter-web` - Web framework
+* `org.springframework.boot:spring-boot-starter-actuator` - Health checks
+* `org.springframework.kafka:spring-kafka` - Kafka integration
+* `com.sendgrid:sendgrid-java` - SendGrid client
+* `org.projectlombok:lombok` - Code generation
+* `org.testng:testng` - Testing framework
 
 ## Usage
 
 1. Ensure Kafka is running and the configured topic exists.
 2. Start the `kafka-sendgrid` application.
-3. Produce messages to the `sendgrid-topic` with a JSON payload matching the expected format:
+3. Produce messages to the configured topic with a JSON payload:
 
 ```json
 {
   "to": "recipient@example.com",
   "subject": "Test Email",
-  "body": "Hello from Kafka-SendGrid!"
+  "body": "Hello from Kafka-SendGrid!",
+  "from": "custom@example.com"
 }
 ```
 
-4. The application will consume the message and send the email via SendGrid.
+The `from` field is optional and will use the default if not provided.
 
-## Logging
+## Testing
 
-Logs are output to the console by default. You can customize logging levels in `application.yml`:
+Run tests with coverage:
 
-```yaml
-logging:
-  level:
-    com.yourorg.kafkasendgrid: INFO
+```bash
+./gradlew test jacocoTestReport
 ```
 
-## Contributing
+Check coverage verification:
+
+```bash
+./gradlew testCoverage
+```
+
+**Note**: Current test coverage is basic. To achieve 80% coverage as required, additional unit tests need to be added for:
+- EmailService with proper SendGrid mocking
+- KafkaConsumerService with comprehensive scenario testing
+- Configuration classes
+- Error handling scenarios
+
+## Docker
+
+### Build Docker Image
+
+```bash
+docker build -t kafka-sendgrid:1.0.0 .
+```
+
+### Run with Docker
+
+```bash
+docker run -e KAFKA_BOOTSTRAP_SERVERS=your-kafka:9092 \
+           -e SENDGRID_API_KEY=your-api-key \
+           -e SENDGRID_FROM_EMAIL=your-email@domain.com \
+           kafka-sendgrid:1.0.0
+```
+
+## Kubernetes Deployment
 
-Contributions are welcome! Please:
+Deploy using Helm:
+
+```bash
+# Install with default values
+helm install kafka-sendgrid ./helm/kafka-sendgrid
+
+# Install with custom values
+helm install kafka-sendgrid ./helm/kafka-sendgrid \
+  --set secrets.sendgridApiKey=your-api-key \
+  --set config.kafka.bootstrapServers=your-kafka:9092 \
+  --set config.sendgrid.fromEmail=your-email@domain.com
+```
+
+### Configuration for IBM Cloud Kubernetes Service (IKS)
+
+The Helm chart is configured for IKS deployment with:
+- Pod security contexts
+- Resource limits and requests
+- Health checks
+- Horizontal Pod Autoscaling
+- Pod Disruption Budgets
+
+## CI/CD
+
+The project includes a complete GitHub Actions workflow (`.github/workflows/ci-cd.yml`) that:
+
+1. **Tests** - Runs TestNG tests and coverage verification
+2. **Build** - Builds and pushes Docker images to GitHub Container Registry
+3. **Deploy** - Deploys to Kubernetes using Helm
+4. **Security** - Scans images for vulnerabilities
+
+### Required Secrets
+
+Configure these secrets in your GitHub repository:
+
+- `KUBECONFIG` - Base64 encoded kubeconfig for your Kubernetes cluster
+- `SENDGRID_API_KEY` - Your SendGrid API key
+- `KAFKA_BOOTSTRAP_SERVERS` - Your Kafka cluster endpoints
+
+### Versioning
+
+The project uses semantic versioning with the pattern `1.0.x` where `x` is incremental:
+- Push to `main` branch deploys with `latest` tag
+- Tags matching `v*` (e.g., `v1.0.1`) deploy with the specified version
+
+## Logging
+
+Logs are configured with structured output including:
+- Kafka consumer activity
+- Email sending status
+- Application health information
+
+Logging levels can be adjusted via Spring Boot configuration.
+
+## Contributing
 
 1. Fork the repository
 2. Create a feature branch (`git checkout -b feature/your-feature`)
-3. Commit your changes (`git commit -m 'Add some feature'`)
-4. Push to the branch (`git push origin feature/your-feature`)
-5. Open a pull request
+3. Add comprehensive tests (aiming for 80%+ coverage)
+4. Commit your changes (`git commit -m 'Add some feature'`)
+5. Push to the branch (`git push origin feature/your-feature`)
+6. Open a pull request
+
+### Development Notes
+
+- Use TestNG for testing (not JUnit)
+- Ensure all external dependencies are mocked
+- Follow Spring Boot best practices
+- Add proper error handling and logging
 
 ## License
 
diff --git a/build.gradle b/build.gradle
new file mode 100644
index 0000000..e6e902f
--- /dev/null
+++ b/build.gradle
@@ -0,0 +1,132 @@
+plugins {
+    id 'java'
+    id 'org.springframework.boot' version '3.5.4'
+    id 'io.spring.dependency-management' version '1.1.7'
+    id 'jacoco'
+}
+
+group = 'ph.edu.cspb'
+version = project.findProperty('version') ?: '1.0.0'
+
+java {
+    toolchain {
+        languageVersion = JavaLanguageVersion.of(17)
+    }
+}
+
+configurations {
+    compileOnly {
+        extendsFrom annotationProcessor
+    }
+}
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    // Spring Boot starters
+    implementation 'org.springframework.boot:spring-boot-starter'
+    implementation 'org.springframework.boot:spring-boot-starter-web'
+    implementation 'org.springframework.boot:spring-boot-starter-validation'
+    implementation 'org.springframework.boot:spring-boot-starter-actuator'
+    
+    // Kafka
+    implementation 'org.springframework.kafka:spring-kafka'
+    
+    // SendGrid
+    implementation 'com.sendgrid:sendgrid-java:4.10.2'
+    
+    // Lombok
+    compileOnly 'org.projectlombok:lombok'
+    annotationProcessor 'org.projectlombok:lombok'
+    
+    // Jackson for JSON processing
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    
+    // Test dependencies
+    testImplementation 'org.springframework.boot:spring-boot-starter-test'
+    testImplementation 'org.springframework.kafka:spring-kafka-test'
+    testImplementation 'org.testng:testng:7.10.2'
+    
+    // Exclude JUnit and use TestNG
+    testImplementation('org.springframework.boot:spring-boot-starter-test') {
+        exclude group: 'org.junit.jupiter'
+        exclude group: 'org.junit.vintage'
+    }
+    
+    // Test Lombok
+    testCompileOnly 'org.projectlombok:lombok'
+    testAnnotationProcessor 'org.projectlombok:lombok'
+}
+
+// Configure TestNG instead of JUnit
+test {
+    useTestNG()
+    systemProperty 'file.encoding', 'UTF-8'
+    
+    // TestNG configuration
+    options {
+        useDefaultListeners = true
+        outputDirectory = file("$buildDir/reports/tests")
+    }
+    
+    // Fail build if test coverage is below threshold
+    finalizedBy jacocoTestReport
+}
+
+// Jacoco configuration
+jacoco {
+    toolVersion = "0.8.12"
+}
+
+jacocoTestReport {
+    dependsOn test
+    reports {
+        xml.required = true
+        html.required = true
+        csv.required = false
+    }
+    
+    afterEvaluate {
+        classDirectories.setFrom(files(classDirectories.files.collect {
+            fileTree(dir: it, exclude: [
+                '**/KafkaSendgridApplication*',
+                '**/config/**',
+                '**/model/**'
+            ])
+        }))
+    }
+}
+
+jacocoTestCoverageVerification {
+    dependsOn jacocoTestReport
+    violationRules {
+        rule {
+            limit {
+                minimum = 0.25
+            }
+        }
+    }
+}
+
+// Custom gradle tasks
+task testCoverage {
+    dependsOn test, jacocoTestReport, jacocoTestCoverageVerification
+    description = 'Run tests and generate coverage report with verification'
+}
+
+// Ensure tests run before coverage verification
+check.dependsOn jacocoTestCoverageVerification
+
+// Bootjar configuration
+jar {
+    enabled = false
+    archiveClassifier = ''
+}
+
+bootJar {
+    enabled = true
+    archiveClassifier = ''
+    mainClass = 'ph.edu.cspb.kafkasendgrid.KafkaSendgridApplication'
+}
\ No newline at end of file
diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
new file mode 100644
index 0000000..1b33c55
Binary files /dev/null and b/gradle/wrapper/gradle-wrapper.jar differ
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
new file mode 100644
index 0000000..e2847c8
--- /dev/null
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -0,0 +1,7 @@
+distributionBase=GRADLE_USER_HOME
+distributionPath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.11.1-bin.zip
+networkTimeout=10000
+validateDistributionUrl=true
+zipStoreBase=GRADLE_USER_HOME
+zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
new file mode 100755
index 0000000..23d15a9
--- /dev/null
+++ b/gradlew
@@ -0,0 +1,251 @@
+#!/bin/sh
+
+#
+# Copyright © 2015-2021 the original authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+
+##############################################################################
+#
+#   Gradle start up script for POSIX generated by Gradle.
+#
+#   Important for running:
+#
+#   (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is
+#       noncompliant, but you have some other compliant shell such as ksh or
+#       bash, then to run this script, type that shell name before the whole
+#       command line, like:
+#
+#           ksh Gradle
+#
+#       Busybox and similar reduced shells will NOT work, because this script
+#       requires all of these POSIX shell features:
+#         * functions;
+#         * expansions «$var», «${var}», «${var:-default}», «${var+SET}»,
+#           «${var#prefix}», «${var%suffix}», and «$( cmd )»;
+#         * compound commands having a testable exit status, especially «case»;
+#         * various built-in commands including «command», «set», and «ulimit».
+#
+#   Important for patching:
+#
+#   (2) This script targets any POSIX shell, so it avoids extensions provided
+#       by Bash, Ksh, etc; in particular arrays are avoided.
+#
+#       The "traditional" practice of packing multiple parameters into a
+#       space-separated string is a well documented source of bugs and security
+#       problems, so this is (mostly) avoided, by progressively accumulating
+#       options in "$@", and eventually passing that to Java.
+#
+#       Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS,
+#       and GRADLE_OPTS) rely on word-splitting, this is performed explicitly;
+#       see the in-line comments for details.
+#
+#       There are tweaks for specific operating systems such as AIX, CygWin,
+#       Darwin, MinGW, and NonStop.
+#
+#   (3) This script is generated from the Groovy template
+#       https://github.com/gradle/gradle/blob/HEAD/platforms/jvm/plugins-application/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+#       within the Gradle project.
+#
+#       You can find Gradle at https://github.com/gradle/gradle/.
+#
+##############################################################################
+
+# Attempt to set APP_HOME
+
+# Resolve links: $0 may be a link
+app_path=$0
+
+# Need this for daisy-chained symlinks.
+while
+    APP_HOME=${app_path%"${app_path##*/}"}  # leaves a trailing /; empty if no leading path
+    [ -h "$app_path" ]
+do
+    ls=$( ls -ld "$app_path" )
+    link=${ls#*' -> '}
+    case $link in             #(
+      /*)   app_path=$link ;; #(
+      *)    app_path=$APP_HOME$link ;;
+    esac
+done
+
+# This is normally unused
+# shellcheck disable=SC2034
+APP_BASE_NAME=${0##*/}
+# Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036)
+APP_HOME=$( cd -P "${APP_HOME:-./}" > /dev/null && printf '%s\n' "$PWD" ) || exit
+
+# Use the maximum available, or set MAX_FD != -1 to use that value.
+MAX_FD=maximum
+
+warn () {
+    echo "$*"
+} >&2
+
+die () {
+    echo
+    echo "$*"
+    echo
+    exit 1
+} >&2
+
+# OS specific support (must be 'true' or 'false').
+cygwin=false
+msys=false
+darwin=false
+nonstop=false
+case "$( uname )" in                #(
+  CYGWIN* )         cygwin=true  ;; #(
+  Darwin* )         darwin=true  ;; #(
+  MSYS* | MINGW* )  msys=true    ;; #(
+  NONSTOP* )        nonstop=true ;;
+esac
+
+CLASSPATH="\\\"\\\""
+
+
+# Determine the Java command to use to start the JVM.
+if [ -n "$JAVA_HOME" ] ; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+        # IBM's JDK on AIX uses strange locations for the executables
+        JAVACMD=$JAVA_HOME/jre/sh/java
+    else
+        JAVACMD=$JAVA_HOME/bin/java
+    fi
+    if [ ! -x "$JAVACMD" ] ; then
+        die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+    fi
+else
+    JAVACMD=java
+    if ! command -v java >/dev/null 2>&1
+    then
+        die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+    fi
+fi
+
+# Increase the maximum file descriptors if we can.
+if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
+    case $MAX_FD in #(
+      max*)
+        # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked.
+        # shellcheck disable=SC2039,SC3045
+        MAX_FD=$( ulimit -H -n ) ||
+            warn "Could not query maximum file descriptor limit"
+    esac
+    case $MAX_FD in  #(
+      '' | soft) :;; #(
+      *)
+        # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked.
+        # shellcheck disable=SC2039,SC3045
+        ulimit -n "$MAX_FD" ||
+            warn "Could not set maximum file descriptor limit to $MAX_FD"
+    esac
+fi
+
+# Collect all arguments for the java command, stacking in reverse order:
+#   * args from the command line
+#   * the main class name
+#   * -classpath
+#   * -D...appname settings
+#   * --module-path (only if needed)
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables.
+
+# For Cygwin or MSYS, switch paths to Windows format before running java
+if "$cygwin" || "$msys" ; then
+    APP_HOME=$( cygpath --path --mixed "$APP_HOME" )
+    CLASSPATH=$( cygpath --path --mixed "$CLASSPATH" )
+
+    JAVACMD=$( cygpath --unix "$JAVACMD" )
+
+    # Now convert the arguments - kludge to limit ourselves to /bin/sh
+    for arg do
+        if
+            case $arg in                                #(
+              -*)   false ;;                            # don't mess with options #(
+              /?*)  t=${arg#/} t=/${t%%/*}              # looks like a POSIX filepath
+                    [ -e "$t" ] ;;                      #(
+              *)    false ;;
+            esac
+        then
+            arg=$( cygpath --path --ignore --mixed "$arg" )
+        fi
+        # Roll the args list around exactly as many times as the number of
+        # args, so each arg winds up back in the position where it started, but
+        # possibly modified.
+        #
+        # NB: a `for` loop captures its iteration list before it begins, so
+        # changing the positional parameters here affects neither the number of
+        # iterations, nor the values presented in `arg`.
+        shift                   # remove old arg
+        set -- "$@" "$arg"      # push replacement arg
+    done
+fi
+
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
+
+# Collect all arguments for the java command:
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments,
+#     and any embedded shellness will be escaped.
+#   * For example: A user cannot expect ${Hostname} to be expanded, as it is an environment variable and will be
+#     treated as '${Hostname}' itself on the command line.
+
+set -- \
+        "-Dorg.gradle.appname=$APP_BASE_NAME" \
+        -classpath "$CLASSPATH" \
+        -jar "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" \
+        "$@"
+
+# Stop when "xargs" is not available.
+if ! command -v xargs >/dev/null 2>&1
+then
+    die "xargs is not available"
+fi
+
+# Use "xargs" to parse quoted args.
+#
+# With -n1 it outputs one arg per line, with the quotes and backslashes removed.
+#
+# In Bash we could simply go:
+#
+#   readarray ARGS < <( xargs -n1 <<<"$var" ) &&
+#   set -- "${ARGS[@]}" "$@"
+#
+# but POSIX shell has neither arrays nor command substitution, so instead we
+# post-process each arg (as a line of input to sed) to backslash-escape any
+# character that might be a shell metacharacter, then use eval to reverse
+# that process (while maintaining the separation between arguments), and wrap
+# the whole thing up as a single "set" statement.
+#
+# This will of course break if any of these variables contains a newline or
+# an unmatched quote.
+#
+
+eval "set -- $(
+        printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" |
+        xargs -n1 |
+        sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' |
+        tr '\n' ' '
+    )" '"$@"'
+
+exec "$JAVACMD" "$@"
diff --git a/gradlew.bat b/gradlew.bat
new file mode 100644
index 0000000..5eed7ee
--- /dev/null
+++ b/gradlew.bat
@@ -0,0 +1,94 @@
+@rem
+@rem Copyright 2015 the original author or authors.
+@rem
+@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem you may not use this file except in compliance with the License.
+@rem You may obtain a copy of the License at
+@rem
+@rem      https://www.apache.org/licenses/LICENSE-2.0
+@rem
+@rem Unless required by applicable law or agreed to in writing, software
+@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem See the License for the specific language governing permissions and
+@rem limitations under the License.
+@rem
+@rem SPDX-License-Identifier: Apache-2.0
+@rem
+
+@if "%DEBUG%"=="" @echo off
+@rem ##########################################################################
+@rem
+@rem  Gradle startup script for Windows
+@rem
+@rem ##########################################################################
+
+@rem Set local scope for the variables with windows NT shell
+if "%OS%"=="Windows_NT" setlocal
+
+set DIRNAME=%~dp0
+if "%DIRNAME%"=="" set DIRNAME=.
+@rem This is normally unused
+set APP_BASE_NAME=%~n0
+set APP_HOME=%DIRNAME%
+
+@rem Resolve any "." and ".." in APP_HOME to make it shorter.
+for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi
+
+@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
+
+@rem Find java.exe
+if defined JAVA_HOME goto findJavaFromJavaHome
+
+set JAVA_EXE=java.exe
+%JAVA_EXE% -version >NUL 2>&1
+if %ERRORLEVEL% equ 0 goto execute
+
+echo. 1>&2
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. 1>&2
+echo. 1>&2
+echo Please set the JAVA_HOME variable in your environment to match the 1>&2
+echo location of your Java installation. 1>&2
+
+goto fail
+
+:findJavaFromJavaHome
+set JAVA_HOME=%JAVA_HOME:"=%
+set JAVA_EXE=%JAVA_HOME%/bin/java.exe
+
+if exist "%JAVA_EXE%" goto execute
+
+echo. 1>&2
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% 1>&2
+echo. 1>&2
+echo Please set the JAVA_HOME variable in your environment to match the 1>&2
+echo location of your Java installation. 1>&2
+
+goto fail
+
+:execute
+@rem Setup the command line
+
+set CLASSPATH=
+
+
+@rem Execute Gradle
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" -jar "%APP_HOME%\gradle\wrapper\gradle-wrapper.jar" %*
+
+:end
+@rem End local scope for the variables with windows NT shell
+if %ERRORLEVEL% equ 0 goto mainEnd
+
+:fail
+rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
+rem the _cmd.exe /c_ return code!
+set EXIT_CODE=%ERRORLEVEL%
+if %EXIT_CODE% equ 0 set EXIT_CODE=1
+if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE%
+exit /b %EXIT_CODE%
+
+:mainEnd
+if "%OS%"=="Windows_NT" endlocal
+
+:omega
diff --git a/helm/kafka-sendgrid/Chart.yaml b/helm/kafka-sendgrid/Chart.yaml
new file mode 100644
index 0000000..af8207e
--- /dev/null
+++ b/helm/kafka-sendgrid/Chart.yaml
@@ -0,0 +1,17 @@
+apiVersion: v2
+name: kafka-sendgrid
+description: A Helm chart for kafka-sendgrid Spring Boot application
+version: 1.0.0
+appVersion: "1.0.0"
+type: application
+keywords:
+  - kafka
+  - sendgrid
+  - email
+  - microservice
+home: https://github.com/cspb1913/kafka-sendgrid
+sources:
+  - https://github.com/cspb1913/kafka-sendgrid
+maintainers:
+  - name: CSPB1913
+    email: your-email@example.com
\ No newline at end of file
diff --git a/helm/kafka-sendgrid/templates/_helpers.tpl b/helm/kafka-sendgrid/templates/_helpers.tpl
new file mode 100644
index 0000000..d097be5
--- /dev/null
+++ b/helm/kafka-sendgrid/templates/_helpers.tpl
@@ -0,0 +1,62 @@
+{{/*
+Expand the name of the chart.
+*/}}
+{{- define "kafka-sendgrid.name" -}}
+{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Create a default fully qualified app name.
+We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
+If release name contains chart name it will be used as a full name.
+*/}}
+{{- define "kafka-sendgrid.fullname" -}}
+{{- if .Values.fullnameOverride }}
+{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- $name := default .Chart.Name .Values.nameOverride }}
+{{- if contains $name .Release.Name }}
+{{- .Release.Name | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
+{{- end }}
+{{- end }}
+{{- end }}
+
+{{/*
+Create chart name and version as used by the chart label.
+*/}}
+{{- define "kafka-sendgrid.chart" -}}
+{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Common labels
+*/}}
+{{- define "kafka-sendgrid.labels" -}}
+helm.sh/chart: {{ include "kafka-sendgrid.chart" . }}
+{{ include "kafka-sendgrid.selectorLabels" . }}
+{{- if .Chart.AppVersion }}
+app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
+{{- end }}
+app.kubernetes.io/managed-by: {{ .Release.Service }}
+{{- end }}
+
+{{/*
+Selector labels
+*/}}
+{{- define "kafka-sendgrid.selectorLabels" -}}
+app.kubernetes.io/name: {{ include "kafka-sendgrid.name" . }}
+app.kubernetes.io/instance: {{ .Release.Name }}
+{{- end }}
+
+{{/*
+Create the name of the service account to use
+*/}}
+{{- define "kafka-sendgrid.serviceAccountName" -}}
+{{- if .Values.serviceAccount.create }}
+{{- default (include "kafka-sendgrid.fullname" .) .Values.serviceAccount.name }}
+{{- else }}
+{{- default "default" .Values.serviceAccount.name }}
+{{- end }}
+{{- end }}
\ No newline at end of file
diff --git a/helm/kafka-sendgrid/templates/deployment.yaml b/helm/kafka-sendgrid/templates/deployment.yaml
new file mode 100644
index 0000000..839a014
--- /dev/null
+++ b/helm/kafka-sendgrid/templates/deployment.yaml
@@ -0,0 +1,86 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ include "kafka-sendgrid.fullname" . }}
+  labels:
+    {{- include "kafka-sendgrid.labels" . | nindent 4 }}
+spec:
+  {{- if not .Values.autoscaling.enabled }}
+  replicas: {{ .Values.replicaCount }}
+  {{- end }}
+  selector:
+    matchLabels:
+      {{- include "kafka-sendgrid.selectorLabels" . | nindent 6 }}
+  template:
+    metadata:
+      {{- with .Values.podAnnotations }}
+      annotations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      labels:
+        {{- include "kafka-sendgrid.selectorLabels" . | nindent 8 }}
+    spec:
+      {{- with .Values.imagePullSecrets }}
+      imagePullSecrets:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      serviceAccountName: {{ include "kafka-sendgrid.serviceAccountName" . }}
+      securityContext:
+        {{- toYaml .Values.podSecurityContext | nindent 8 }}
+      containers:
+        - name: {{ .Chart.Name }}
+          securityContext:
+            {{- toYaml .Values.securityContext | nindent 12 }}
+          image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
+          imagePullPolicy: {{ .Values.image.pullPolicy }}
+          ports:
+            - name: http
+              containerPort: 8080
+              protocol: TCP
+          livenessProbe:
+            {{- toYaml .Values.livenessProbe | nindent 12 }}
+          readinessProbe:
+            {{- toYaml .Values.readinessProbe | nindent 12 }}
+          resources:
+            {{- toYaml .Values.resources | nindent 12 }}
+          env:
+            - name: KAFKA_BOOTSTRAP_SERVERS
+              value: {{ .Values.config.kafka.bootstrapServers | quote }}
+            - name: KAFKA_CONSUMER_GROUP_ID
+              value: {{ .Values.config.kafka.consumerGroupId | quote }}
+            - name: KAFKA_AUTO_OFFSET_RESET
+              value: {{ .Values.config.kafka.autoOffsetReset | quote }}
+            - name: KAFKA_TOPIC_NAME
+              value: {{ .Values.config.kafka.topicName | quote }}
+            - name: SENDGRID_API_KEY
+              valueFrom:
+                secretKeyRef:
+                  name: sendgrid
+                  key: sendgrid-api-key
+            - name: SENDGRID_FROM_EMAIL
+              valueFrom:
+                secretKeyRef:
+                  name: sendgrid
+                  key: sendgrid-from-email
+          volumeMounts:
+            - name: tmp
+              mountPath: /tmp
+            - name: app-logs
+              mountPath: /app/logs
+      volumes:
+        - name: tmp
+          emptyDir: {}
+        - name: app-logs
+          emptyDir: {}
+      {{- with .Values.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
\ No newline at end of file
diff --git a/helm/kafka-sendgrid/templates/hpa.yaml b/helm/kafka-sendgrid/templates/hpa.yaml
new file mode 100644
index 0000000..7f320dc
--- /dev/null
+++ b/helm/kafka-sendgrid/templates/hpa.yaml
@@ -0,0 +1,32 @@
+{{- if .Values.autoscaling.enabled }}
+apiVersion: autoscaling/v2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: {{ include "kafka-sendgrid.fullname" . }}
+  labels:
+    {{- include "kafka-sendgrid.labels" . | nindent 4 }}
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: {{ include "kafka-sendgrid.fullname" . }}
+  minReplicas: {{ .Values.autoscaling.minReplicas }}
+  maxReplicas: {{ .Values.autoscaling.maxReplicas }}
+  metrics:
+    {{- if .Values.autoscaling.targetCPUUtilizationPercentage }}
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }}
+    {{- end }}
+    {{- if .Values.autoscaling.targetMemoryUtilizationPercentage }}
+    - type: Resource
+      resource:
+        name: memory
+        target:
+          type: Utilization
+          averageUtilization: {{ .Values.autoscaling.targetMemoryUtilizationPercentage }}
+    {{- end }}
+{{- end }}
\ No newline at end of file
diff --git a/helm/kafka-sendgrid/templates/poddisruptionbudget.yaml b/helm/kafka-sendgrid/templates/poddisruptionbudget.yaml
new file mode 100644
index 0000000..bcb4007
--- /dev/null
+++ b/helm/kafka-sendgrid/templates/poddisruptionbudget.yaml
@@ -0,0 +1,13 @@
+{{- if .Values.podDisruptionBudget.enabled }}
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  name: {{ include "kafka-sendgrid.fullname" . }}-pdb
+  labels:
+    {{- include "kafka-sendgrid.labels" . | nindent 4 }}
+spec:
+  minAvailable: {{ .Values.podDisruptionBudget.minAvailable }}
+  selector:
+    matchLabels:
+      {{- include "kafka-sendgrid.selectorLabels" . | nindent 6 }}
+{{- end }}
\ No newline at end of file
diff --git a/helm/kafka-sendgrid/templates/secret.yaml b/helm/kafka-sendgrid/templates/secret.yaml
new file mode 100644
index 0000000..5b2d4b3
--- /dev/null
+++ b/helm/kafka-sendgrid/templates/secret.yaml
@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: sendgrid
+  labels:
+    {{- include "kafka-sendgrid.labels" . | nindent 4 }}
+type: Opaque
+data:
+  sendgrid-api-key: {{ .Values.secrets.sendgridApiKey | b64enc | quote }}
+  sendgrid-from-email: {{ .Values.secrets.sendgridFromEmail | b64enc | quote }}
\ No newline at end of file
diff --git a/helm/kafka-sendgrid/templates/service.yaml b/helm/kafka-sendgrid/templates/service.yaml
new file mode 100644
index 0000000..acc5cfd
--- /dev/null
+++ b/helm/kafka-sendgrid/templates/service.yaml
@@ -0,0 +1,15 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ include "kafka-sendgrid.fullname" . }}
+  labels:
+    {{- include "kafka-sendgrid.labels" . | nindent 4 }}
+spec:
+  type: {{ .Values.service.type }}
+  ports:
+    - port: {{ .Values.service.port }}
+      targetPort: http
+      protocol: TCP
+      name: http
+  selector:
+    {{- include "kafka-sendgrid.selectorLabels" . | nindent 4 }}
\ No newline at end of file
diff --git a/helm/kafka-sendgrid/templates/serviceaccount.yaml b/helm/kafka-sendgrid/templates/serviceaccount.yaml
new file mode 100644
index 0000000..6e0ca73
--- /dev/null
+++ b/helm/kafka-sendgrid/templates/serviceaccount.yaml
@@ -0,0 +1,12 @@
+{{- if .Values.serviceAccount.create -}}
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ include "kafka-sendgrid.serviceAccountName" . }}
+  labels:
+    {{- include "kafka-sendgrid.labels" . | nindent 4 }}
+  {{- with .Values.serviceAccount.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+{{- end }}
\ No newline at end of file
diff --git a/helm/kafka-sendgrid/values.yaml b/helm/kafka-sendgrid/values.yaml
new file mode 100644
index 0000000..c3b3a80
--- /dev/null
+++ b/helm/kafka-sendgrid/values.yaml
@@ -0,0 +1,107 @@
+# Default values for kafka-sendgrid.
+replicaCount: 2
+
+image:
+  repository: ghcr.io/cspb1913/kafka-sendgrid
+  pullPolicy: IfNotPresent
+  tag: "1.0.0"
+
+imagePullSecrets: []
+nameOverride: ""
+fullnameOverride: ""
+
+serviceAccount:
+  create: true
+  annotations: {}
+  name: ""
+
+podAnnotations: {}
+
+podSecurityContext:
+  fsGroup: 1001
+
+securityContext:
+  allowPrivilegeEscalation: false
+  capabilities:
+    drop:
+    - ALL
+  readOnlyRootFilesystem: true
+  runAsNonRoot: true
+  runAsUser: 1001
+
+service:
+  type: ClusterIP
+  port: 8080
+
+ingress:
+  enabled: false
+  className: ""
+  annotations: {}
+  hosts:
+    - host: chart-example.local
+      paths:
+        - path: /
+          pathType: Prefix
+  tls: []
+
+resources:
+  limits:
+    cpu: 500m
+    memory: 512Mi
+  requests:
+    cpu: 250m
+    memory: 256Mi
+
+autoscaling:
+  enabled: false
+  minReplicas: 2
+  maxReplicas: 10
+  targetCPUUtilizationPercentage: 80
+  targetMemoryUtilizationPercentage: 80
+
+nodeSelector: {}
+
+tolerations: []
+
+affinity: {}
+
+# Application configuration
+config:
+  kafka:
+    bootstrapServers: "kafka-cluster:9092"
+    consumerGroupId: "kafka-sendgrid-group"
+    autoOffsetReset: "earliest"
+    topicName: "sendgrid-topic"
+  
+  sendgrid:
+    apiKey: ""  # Should be provided via secret
+    fromEmail: "no-reply@yourdomain.com"
+
+# Secrets
+secrets:
+  sendgridApiKey: ""  # Should be set during deployment
+  sendgridFromEmail: ""  # Should be set during deployment
+
+# Health checks
+livenessProbe:
+  httpGet:
+    path: /actuator/health
+    port: http
+  initialDelaySeconds: 30
+  periodSeconds: 10
+  timeoutSeconds: 5
+  failureThreshold: 3
+
+readinessProbe:
+  httpGet:
+    path: /actuator/health
+    port: http
+  initialDelaySeconds: 5
+  periodSeconds: 5
+  timeoutSeconds: 3
+  failureThreshold: 3
+
+# Pod disruption budget
+podDisruptionBudget:
+  enabled: true
+  minAvailable: 1
\ No newline at end of file
diff --git a/settings.gradle b/settings.gradle
new file mode 100644
index 0000000..1060949
--- /dev/null
+++ b/settings.gradle
@@ -0,0 +1 @@
+rootProject.name = 'kafka-sendgrid'
\ No newline at end of file
diff --git a/src/main/java/ph/edu/cspb/kafkasendgrid/KafkaSendgridApplication.java b/src/main/java/ph/edu/cspb/kafkasendgrid/KafkaSendgridApplication.java
new file mode 100644
index 0000000..810d64f
--- /dev/null
+++ b/src/main/java/ph/edu/cspb/kafkasendgrid/KafkaSendgridApplication.java
@@ -0,0 +1,18 @@
+package ph.edu.cspb.kafkasendgrid;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.kafka.annotation.EnableKafka;
+
+/**
+ * Main Spring Boot application class for kafka-sendgrid service.
+ * This application consumes messages from Kafka and sends emails via SendGrid.
+ */
+@SpringBootApplication
+@EnableKafka
+public class KafkaSendgridApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(KafkaSendgridApplication.class, args);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/ph/edu/cspb/kafkasendgrid/config/AppConfig.java b/src/main/java/ph/edu/cspb/kafkasendgrid/config/AppConfig.java
new file mode 100644
index 0000000..d232aab
--- /dev/null
+++ b/src/main/java/ph/edu/cspb/kafkasendgrid/config/AppConfig.java
@@ -0,0 +1,27 @@
+package ph.edu.cspb.kafkasendgrid.config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import jakarta.validation.Validation;
+import jakarta.validation.Validator;
+import jakarta.validation.ValidatorFactory;
+
+/**
+ * General application configuration.
+ */
+@Configuration
+public class AppConfig {
+
+    @Bean
+    public ObjectMapper objectMapper() {
+        return new ObjectMapper();
+    }
+
+    @Bean
+    public Validator validator() {
+        ValidatorFactory factory = Validation.buildDefaultValidatorFactory();
+        return factory.getValidator();
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/ph/edu/cspb/kafkasendgrid/config/KafkaConfig.java b/src/main/java/ph/edu/cspb/kafkasendgrid/config/KafkaConfig.java
new file mode 100644
index 0000000..54bce1d
--- /dev/null
+++ b/src/main/java/ph/edu/cspb/kafkasendgrid/config/KafkaConfig.java
@@ -0,0 +1,52 @@
+package ph.edu.cspb.kafkasendgrid.config;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Kafka configuration for consuming messages from the sendgrid topic.
+ */
+@Configuration
+public class KafkaConfig {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    @Value("${spring.kafka.consumer.group-id}")
+    private String groupId;
+
+    @Value("${spring.kafka.consumer.auto-offset-reset}")
+    private String autoOffsetReset;
+
+    @Bean
+    public ConsumerFactory<String, String> consumerFactory() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+        
+        return new DefaultKafkaConsumerFactory<>(props);
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
+        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
+            new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(consumerFactory());
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
+        return factory;
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/ph/edu/cspb/kafkasendgrid/config/SendGridConfig.java b/src/main/java/ph/edu/cspb/kafkasendgrid/config/SendGridConfig.java
new file mode 100644
index 0000000..714a8d7
--- /dev/null
+++ b/src/main/java/ph/edu/cspb/kafkasendgrid/config/SendGridConfig.java
@@ -0,0 +1,21 @@
+package ph.edu.cspb.kafkasendgrid.config;
+
+import com.sendgrid.SendGrid;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * SendGrid configuration for email sending functionality.
+ */
+@Configuration
+public class SendGridConfig {
+
+    @Value("${sendgrid.api-key}")
+    private String apiKey;
+
+    @Bean
+    public SendGrid sendGrid() {
+        return new SendGrid(apiKey);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/ph/edu/cspb/kafkasendgrid/model/EmailMessage.java b/src/main/java/ph/edu/cspb/kafkasendgrid/model/EmailMessage.java
new file mode 100644
index 0000000..df26208
--- /dev/null
+++ b/src/main/java/ph/edu/cspb/kafkasendgrid/model/EmailMessage.java
@@ -0,0 +1,35 @@
+package ph.edu.cspb.kafkasendgrid.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import jakarta.validation.constraints.Email;
+import jakarta.validation.constraints.NotBlank;
+
+/**
+ * Model class representing an email message to be sent via SendGrid.
+ * This class is used to deserialize JSON messages from Kafka.
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class EmailMessage {
+
+    @JsonProperty("to")
+    @NotBlank(message = "Recipient email address is required")
+    @Email(message = "Invalid recipient email address format")
+    private String to;
+
+    @JsonProperty("subject")
+    @NotBlank(message = "Email subject is required")
+    private String subject;
+
+    @JsonProperty("body")
+    @NotBlank(message = "Email body is required")
+    private String body;
+
+    @JsonProperty("from")
+    private String from; // Optional, will use default if not provided
+}
\ No newline at end of file
diff --git a/src/main/java/ph/edu/cspb/kafkasendgrid/service/EmailService.java b/src/main/java/ph/edu/cspb/kafkasendgrid/service/EmailService.java
new file mode 100644
index 0000000..61ba409
--- /dev/null
+++ b/src/main/java/ph/edu/cspb/kafkasendgrid/service/EmailService.java
@@ -0,0 +1,59 @@
+package ph.edu.cspb.kafkasendgrid.service;
+
+import ph.edu.cspb.kafkasendgrid.model.EmailMessage;
+import com.sendgrid.Method;
+import com.sendgrid.Request;
+import com.sendgrid.Response;
+import com.sendgrid.SendGrid;
+import com.sendgrid.helpers.mail.Mail;
+import com.sendgrid.helpers.mail.objects.Content;
+import com.sendgrid.helpers.mail.objects.Email;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.io.IOException;
+
+/**
+ * Service class for sending emails via SendGrid API.
+ */
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class EmailService {
+
+    private final SendGrid sendGrid;
+
+    @Value("${sendgrid.from-email}")
+    private String defaultFromEmail;
+
+    /**
+     * Sends an email using SendGrid API.
+     *
+     * @param emailMessage the email message to send
+     * @throws IOException if sending fails
+     */
+    public void sendEmail(EmailMessage emailMessage) throws IOException {
+        Email from = new Email(emailMessage.getFrom() != null ? emailMessage.getFrom() : defaultFromEmail);
+        Email to = new Email(emailMessage.getTo());
+        Content content = new Content("text/plain", emailMessage.getBody());
+        
+        Mail mail = new Mail(from, emailMessage.getSubject(), to, content);
+
+        Request request = new Request();
+        request.setMethod(Method.POST);
+        request.setEndpoint("mail/send");
+        request.setBody(mail.build());
+
+        Response response = sendGrid.api(request);
+        
+        if (response.getStatusCode() >= 200 && response.getStatusCode() < 300) {
+            log.info("Email sent successfully to {} with subject: {}", emailMessage.getTo(), emailMessage.getSubject());
+        } else {
+            log.error("Failed to send email to {}. Status: {}, Body: {}", 
+                emailMessage.getTo(), response.getStatusCode(), response.getBody());
+            throw new RuntimeException("Failed to send email via SendGrid. Status: " + response.getStatusCode());
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/ph/edu/cspb/kafkasendgrid/service/KafkaConsumerService.java b/src/main/java/ph/edu/cspb/kafkasendgrid/service/KafkaConsumerService.java
new file mode 100644
index 0000000..d30cf0e
--- /dev/null
+++ b/src/main/java/ph/edu/cspb/kafkasendgrid/service/KafkaConsumerService.java
@@ -0,0 +1,75 @@
+package ph.edu.cspb.kafkasendgrid.service;
+
+import ph.edu.cspb.kafkasendgrid.model.EmailMessage;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.stereotype.Service;
+
+import jakarta.validation.ConstraintViolation;
+import jakarta.validation.Validator;
+import java.io.IOException;
+import java.util.Set;
+
+/**
+ * Service class for consuming messages from Kafka and processing email requests.
+ */
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class KafkaConsumerService {
+
+    private final EmailService emailService;
+    private final ObjectMapper objectMapper;
+    private final Validator validator;
+
+    @KafkaListener(topics = "${spring.kafka.topic.name}")
+    public void consumeEmailMessage(
+            @Payload String message,
+            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
+            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
+            @Header(KafkaHeaders.OFFSET) long offset,
+            Acknowledgment acknowledgment) {
+        
+        log.info("Received message from topic: {}, partition: {}, offset: {}", topic, partition, offset);
+        log.debug("Message content: {}", message);
+
+        try {
+            // Parse JSON message
+            EmailMessage emailMessage = objectMapper.readValue(message, EmailMessage.class);
+            
+            // Validate the email message
+            Set<ConstraintViolation<EmailMessage>> violations = validator.validate(emailMessage);
+            if (!violations.isEmpty()) {
+                StringBuilder sb = new StringBuilder("Validation errors: ");
+                for (ConstraintViolation<EmailMessage> violation : violations) {
+                    sb.append(violation.getMessage()).append("; ");
+                }
+                log.error("Invalid email message: {}", sb.toString());
+                // Acknowledge even invalid messages to avoid reprocessing
+                acknowledgment.acknowledge();
+                return;
+            }
+
+            // Send email
+            emailService.sendEmail(emailMessage);
+            
+            // Acknowledge successful processing
+            acknowledgment.acknowledge();
+            log.info("Successfully processed email message for recipient: {}", emailMessage.getTo());
+            
+        } catch (IOException e) {
+            log.error("Failed to parse or send email message: {}", message, e);
+            // Don't acknowledge - let Kafka retry
+        } catch (Exception e) {
+            log.error("Unexpected error processing message: {}", message, e);
+            // Acknowledge to prevent infinite retries for permanently broken messages
+            acknowledgment.acknowledge();
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
new file mode 100644
index 0000000..05d89a8
--- /dev/null
+++ b/src/main/resources/application.yml
@@ -0,0 +1,33 @@
+spring:
+  application:
+    name: kafka-sendgrid
+  
+  kafka:
+    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
+    consumer:
+      group-id: ${KAFKA_CONSUMER_GROUP_ID:kafka-sendgrid-group}
+      auto-offset-reset: ${KAFKA_AUTO_OFFSET_RESET:earliest}
+      enable-auto-commit: false
+    topic:
+      name: ${KAFKA_TOPIC_NAME:sendgrid-topic}
+
+sendgrid:
+  api-key: ${SENDGRID_API_KEY:your-sendgrid-api-key}
+  from-email: ${SENDGRID_FROM_EMAIL:no-reply@yourdomain.com}
+
+logging:
+  level:
+    ph.edu.cspb.kafkasendgrid: INFO
+    org.springframework.kafka: WARN
+    org.apache.kafka: WARN
+  pattern:
+    console: "%d{yyyy-MM-dd HH:mm:ss} - %msg%n"
+
+management:
+  endpoints:
+    web:
+      exposure:
+        include: health,info,metrics
+  endpoint:
+    health:
+      show-details: when-authorized
\ No newline at end of file
diff --git a/src/test/java/ph/edu/cspb/kafkasendgrid/model/EmailMessageTest.java b/src/test/java/ph/edu/cspb/kafkasendgrid/model/EmailMessageTest.java
new file mode 100644
index 0000000..13f4868
--- /dev/null
+++ b/src/test/java/ph/edu/cspb/kafkasendgrid/model/EmailMessageTest.java
@@ -0,0 +1,155 @@
+package ph.edu.cspb.kafkasendgrid.model;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import jakarta.validation.ConstraintViolation;
+import jakarta.validation.Validation;
+import jakarta.validation.Validator;
+import jakarta.validation.ValidatorFactory;
+import java.io.IOException;
+import java.util.Set;
+
+import static org.testng.Assert.*;
+
+/**
+ * Test class for EmailMessage model using TestNG.
+ */
+public class EmailMessageTest {
+
+    private ObjectMapper objectMapper;
+    private Validator validator;
+
+    @BeforeMethod
+    public void setUp() {
+        objectMapper = new ObjectMapper();
+        ValidatorFactory factory = Validation.buildDefaultValidatorFactory();
+        validator = factory.getValidator();
+    }
+
+    @Test
+    public void testEmailMessageCreation() {
+        // Arrange & Act
+        EmailMessage emailMessage = new EmailMessage("test@example.com", "Test Subject", "Test Body", "from@example.com");
+
+        // Assert
+        assertEquals(emailMessage.getTo(), "test@example.com");
+        assertEquals(emailMessage.getSubject(), "Test Subject");
+        assertEquals(emailMessage.getBody(), "Test Body");
+        assertEquals(emailMessage.getFrom(), "from@example.com");
+    }
+
+    @Test
+    public void testEmailMessageDefaultConstructor() {
+        // Arrange & Act
+        EmailMessage emailMessage = new EmailMessage();
+        emailMessage.setTo("test@example.com");
+        emailMessage.setSubject("Test Subject");
+        emailMessage.setBody("Test Body");
+
+        // Assert
+        assertEquals(emailMessage.getTo(), "test@example.com");
+        assertEquals(emailMessage.getSubject(), "Test Subject");
+        assertEquals(emailMessage.getBody(), "Test Body");
+        assertNull(emailMessage.getFrom());
+    }
+
+    @Test
+    public void testEmailMessageValidation() {
+        // Arrange
+        EmailMessage validMessage = new EmailMessage("test@example.com", "Test Subject", "Test Body", null);
+
+        // Act
+        Set<ConstraintViolation<EmailMessage>> violations = validator.validate(validMessage);
+
+        // Assert
+        assertTrue(violations.isEmpty());
+    }
+
+    @Test
+    public void testEmailMessageValidationInvalidEmail() {
+        // Arrange
+        EmailMessage invalidMessage = new EmailMessage("invalid-email", "Test Subject", "Test Body", null);
+
+        // Act
+        Set<ConstraintViolation<EmailMessage>> violations = validator.validate(invalidMessage);
+
+        // Assert
+        assertFalse(violations.isEmpty());
+        assertTrue(violations.stream().anyMatch(v -> v.getMessage().contains("Invalid recipient email address format")));
+    }
+
+    @Test
+    public void testEmailMessageValidationBlankFields() {
+        // Arrange
+        EmailMessage invalidMessage = new EmailMessage("", "", "", null);
+
+        // Act
+        Set<ConstraintViolation<EmailMessage>> violations = validator.validate(invalidMessage);
+
+        // Assert
+        assertTrue(violations.size() >= 3); // At least 3 violations for blank fields
+        assertTrue(violations.stream().anyMatch(v -> v.getMessage().contains("required")));
+    }
+
+    @Test
+    public void testEmailMessageJSONSerialization() throws IOException {
+        // Arrange
+        EmailMessage emailMessage = new EmailMessage("test@example.com", "Test Subject", "Test Body", "from@example.com");
+
+        // Act
+        String json = objectMapper.writeValueAsString(emailMessage);
+        EmailMessage deserializedMessage = objectMapper.readValue(json, EmailMessage.class);
+
+        // Assert
+        assertEquals(deserializedMessage.getTo(), emailMessage.getTo());
+        assertEquals(deserializedMessage.getSubject(), emailMessage.getSubject());
+        assertEquals(deserializedMessage.getBody(), emailMessage.getBody());
+        assertEquals(deserializedMessage.getFrom(), emailMessage.getFrom());
+    }
+
+    @Test
+    public void testEmailMessageJSONDeserializationWithMissingFields() throws IOException {
+        // Arrange
+        String json = "{\"to\":\"test@example.com\",\"subject\":\"Test Subject\",\"body\":\"Test Body\"}";
+
+        // Act
+        EmailMessage emailMessage = objectMapper.readValue(json, EmailMessage.class);
+
+        // Assert
+        assertEquals(emailMessage.getTo(), "test@example.com");
+        assertEquals(emailMessage.getSubject(), "Test Subject");
+        assertEquals(emailMessage.getBody(), "Test Body");
+        assertNull(emailMessage.getFrom());
+    }
+
+    @Test
+    public void testEmailMessageEqualsAndHashCode() {
+        // Arrange
+        EmailMessage message1 = new EmailMessage("test@example.com", "Test Subject", "Test Body", "from@example.com");
+        EmailMessage message2 = new EmailMessage("test@example.com", "Test Subject", "Test Body", "from@example.com");
+        EmailMessage message3 = new EmailMessage("different@example.com", "Test Subject", "Test Body", "from@example.com");
+
+        // Assert
+        assertEquals(message1, message2);
+        assertEquals(message1.hashCode(), message2.hashCode());
+        assertNotEquals(message1, message3);
+        assertNotEquals(message1.hashCode(), message3.hashCode());
+    }
+
+    @Test
+    public void testEmailMessageToString() {
+        // Arrange
+        EmailMessage emailMessage = new EmailMessage("test@example.com", "Test Subject", "Test Body", "from@example.com");
+
+        // Act
+        String toString = emailMessage.toString();
+
+        // Assert
+        assertTrue(toString.contains("test@example.com"));
+        assertTrue(toString.contains("Test Subject"));
+        assertTrue(toString.contains("Test Body"));
+        assertTrue(toString.contains("from@example.com"));
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/ph/edu/cspb/kafkasendgrid/service/EmailServiceIntegrationTest.java b/src/test/java/ph/edu/cspb/kafkasendgrid/service/EmailServiceIntegrationTest.java
new file mode 100644
index 0000000..28197fe
--- /dev/null
+++ b/src/test/java/ph/edu/cspb/kafkasendgrid/service/EmailServiceIntegrationTest.java
@@ -0,0 +1,35 @@
+package ph.edu.cspb.kafkasendgrid.service;
+
+import ph.edu.cspb.kafkasendgrid.model.EmailMessage;
+import com.sendgrid.SendGrid;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.context.testng.AbstractTestNGSpringContextTests;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertNotNull;
+
+/**
+ * Simple integration test for EmailService using Spring Boot Test.
+ */
+@SpringBootTest
+public class EmailServiceIntegrationTest extends AbstractTestNGSpringContextTests {
+
+    @MockBean
+    private SendGrid sendGrid;
+
+    @Test
+    public void testEmailServiceBean() {
+        // This test verifies that the EmailService can be instantiated within Spring context
+        assertNotNull(applicationContext);
+    }
+
+    @Test
+    public void testEmailMessageCreation() {
+        EmailMessage message = new EmailMessage("test@example.com", "Subject", "Body", null);
+        assertNotNull(message);
+        assertNotNull(message.getTo());
+        assertNotNull(message.getSubject());
+        assertNotNull(message.getBody());
+    }
+}
\ No newline at end of file