code for kafka consumer

.gitattributes

+/mvnw text eol=lf
+*.cmd text eol=crlf

.gitignore

+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/

.mvn/wrapper/maven-wrapper.properties

+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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
+#
+#   http://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.
+wrapperVersion=3.3.2
+distributionType=only-script
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.9/apache-maven-3.9.9-bin.zip

mvnw.cmd

+<# : batch portion
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM    http://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------
+
+@REM ----------------------------------------------------------------------------
+@REM Apache Maven Wrapper startup batch script, version 3.3.2
+@REM
+@REM Optional ENV vars
+@REM   MVNW_REPOURL - repo url base for downloading maven distribution
+@REM   MVNW_USERNAME/MVNW_PASSWORD - user and password for downloading maven
+@REM   MVNW_VERBOSE - true: enable verbose log; others: silence the output
+@REM ----------------------------------------------------------------------------
+
+@IF "%__MVNW_ARG0_NAME__%"=="" (SET __MVNW_ARG0_NAME__=%~nx0)
+@SET __MVNW_CMD__=
+@SET __MVNW_ERROR__=
+@SET __MVNW_PSMODULEP_SAVE=%PSModulePath%
+@SET PSModulePath=
+@FOR /F "usebackq tokens=1* delims==" %%A IN (`powershell -noprofile "& {$scriptDir='%~dp0'; $script='%__MVNW_ARG0_NAME__%'; icm -ScriptBlock ([Scriptblock]::Create((Get-Content -Raw '%~f0'))) -NoNewScope}"`) DO @(
+  IF "%%A"=="MVN_CMD" (set __MVNW_CMD__=%%B) ELSE IF "%%B"=="" (echo %%A) ELSE (echo %%A=%%B)
+)
+@SET PSModulePath=%__MVNW_PSMODULEP_SAVE%
+@SET __MVNW_PSMODULEP_SAVE=
+@SET __MVNW_ARG0_NAME__=
+@SET MVNW_USERNAME=
+@SET MVNW_PASSWORD=
+@IF NOT "%__MVNW_CMD__%"=="" (%__MVNW_CMD__% %*)
+@echo Cannot start maven from wrapper >&2 && exit /b 1
+@GOTO :EOF
+: end batch / begin powershell #>
+
+$ErrorActionPreference = "Stop"
+if ($env:MVNW_VERBOSE -eq "true") {
+  $VerbosePreference = "Continue"
+}
+
+# calculate distributionUrl, requires .mvn/wrapper/maven-wrapper.properties
+$distributionUrl = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionUrl
+if (!$distributionUrl) {
+  Write-Error "cannot read distributionUrl property in $scriptDir/.mvn/wrapper/maven-wrapper.properties"
+}
+
+switch -wildcard -casesensitive ( $($distributionUrl -replace '^.*/','') ) {
+  "maven-mvnd-*" {
+    $USE_MVND = $true
+    $distributionUrl = $distributionUrl -replace '-bin\.[^.]*$',"-windows-amd64.zip"
+    $MVN_CMD = "mvnd.cmd"
+    break
+  }
+  default {
+    $USE_MVND = $false
+    $MVN_CMD = $script -replace '^mvnw','mvn'
+    break
+  }
+}
+
+# apply MVNW_REPOURL and calculate MAVEN_HOME
+# maven home pattern: ~/.m2/wrapper/dists/{apache-maven-<version>,maven-mvnd-<version>-<platform>}/<hash>
+if ($env:MVNW_REPOURL) {
+  $MVNW_REPO_PATTERN = if ($USE_MVND) { "/org/apache/maven/" } else { "/maven/mvnd/" }
+  $distributionUrl = "$env:MVNW_REPOURL$MVNW_REPO_PATTERN$($distributionUrl -replace '^.*'+$MVNW_REPO_PATTERN,'')"
+}
+$distributionUrlName = $distributionUrl -replace '^.*/',''
+$distributionUrlNameMain = $distributionUrlName -replace '\.[^.]*$','' -replace '-bin$',''
+$MAVEN_HOME_PARENT = "$HOME/.m2/wrapper/dists/$distributionUrlNameMain"
+if ($env:MAVEN_USER_HOME) {
+  $MAVEN_HOME_PARENT = "$env:MAVEN_USER_HOME/wrapper/dists/$distributionUrlNameMain"
+}
+$MAVEN_HOME_NAME = ([System.Security.Cryptography.MD5]::Create().ComputeHash([byte[]][char[]]$distributionUrl) | ForEach-Object {$_.ToString("x2")}) -join ''
+$MAVEN_HOME = "$MAVEN_HOME_PARENT/$MAVEN_HOME_NAME"
+
+if (Test-Path -Path "$MAVEN_HOME" -PathType Container) {
+  Write-Verbose "found existing MAVEN_HOME at $MAVEN_HOME"
+  Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
+  exit $?
+}
+
+if (! $distributionUrlNameMain -or ($distributionUrlName -eq $distributionUrlNameMain)) {
+  Write-Error "distributionUrl is not valid, must end with *-bin.zip, but found $distributionUrl"
+}
+
+# prepare tmp dir
+$TMP_DOWNLOAD_DIR_HOLDER = New-TemporaryFile
+$TMP_DOWNLOAD_DIR = New-Item -Itemtype Directory -Path "$TMP_DOWNLOAD_DIR_HOLDER.dir"
+$TMP_DOWNLOAD_DIR_HOLDER.Delete() | Out-Null
+trap {
+  if ($TMP_DOWNLOAD_DIR.Exists) {
+    try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+    catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+  }
+}
+
+New-Item -Itemtype Directory -Path "$MAVEN_HOME_PARENT" -Force | Out-Null
+
+# Download and Install Apache Maven
+Write-Verbose "Couldn't find MAVEN_HOME, downloading and installing it ..."
+Write-Verbose "Downloading from: $distributionUrl"
+Write-Verbose "Downloading to: $TMP_DOWNLOAD_DIR/$distributionUrlName"
+
+$webclient = New-Object System.Net.WebClient
+if ($env:MVNW_USERNAME -and $env:MVNW_PASSWORD) {
+  $webclient.Credentials = New-Object System.Net.NetworkCredential($env:MVNW_USERNAME, $env:MVNW_PASSWORD)
+}
+[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
+$webclient.DownloadFile($distributionUrl, "$TMP_DOWNLOAD_DIR/$distributionUrlName") | Out-Null
+
+# If specified, validate the SHA-256 sum of the Maven distribution zip file
+$distributionSha256Sum = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionSha256Sum
+if ($distributionSha256Sum) {
+  if ($USE_MVND) {
+    Write-Error "Checksum validation is not supported for maven-mvnd. `nPlease disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties."
+  }
+  Import-Module $PSHOME\Modules\Microsoft.PowerShell.Utility -Function Get-FileHash
+  if ((Get-FileHash "$TMP_DOWNLOAD_DIR/$distributionUrlName" -Algorithm SHA256).Hash.ToLower() -ne $distributionSha256Sum) {
+    Write-Error "Error: Failed to validate Maven distribution SHA-256, your Maven distribution might be compromised. If you updated your Maven version, you need to update the specified distributionSha256Sum property."
+  }
+}
+
+# unzip and move
+Expand-Archive "$TMP_DOWNLOAD_DIR/$distributionUrlName" -DestinationPath "$TMP_DOWNLOAD_DIR" | Out-Null
+Rename-Item -Path "$TMP_DOWNLOAD_DIR/$distributionUrlNameMain" -NewName $MAVEN_HOME_NAME | Out-Null
+try {
+  Move-Item -Path "$TMP_DOWNLOAD_DIR/$MAVEN_HOME_NAME" -Destination $MAVEN_HOME_PARENT | Out-Null
+} catch {
+  if (! (Test-Path -Path "$MAVEN_HOME" -PathType Container)) {
+    Write-Error "fail to move MAVEN_HOME"
+  }
+} finally {
+  try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+  catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+}
+
+Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"

pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>3.4.2</version>
+		<relativePath/> <!-- lookup parent from repository -->
+	</parent>
+	<groupId>com.nisum</groupId>
+	<artifactId>kafka-poc</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<name>kafka-poc</name>
+	<description>Kafka poc</description>
+	<url/>
+	<licenses>
+		<license/>
+	</licenses>
+	<developers>
+		<developer/>
+	</developers>
+	<scm>
+		<connection/>
+		<developerConnection/>
+		<tag/>
+		<url/>
+	</scm>
+	<properties>
+		<java.version>17</java.version>
+	</properties>
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-validation</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<configuration>
+					<annotationProcessorPaths>
+						<path>
+							<groupId>org.projectlombok</groupId>
+							<artifactId>lombok</artifactId>
+						</path>
+					</annotationProcessorPaths>
+				</configuration>
+			</plugin>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+				<configuration>
+					<excludes>
+						<exclude>
+							<groupId>org.projectlombok</groupId>
+							<artifactId>lombok</artifactId>
+						</exclude>
+					</excludes>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>

src/main/java/com/nisum/kafka_poc/KafkaPocApplication.java

+package com.nisum.kafka_poc;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class KafkaPocApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(KafkaPocApplication.class, args);
+	}
+
+}

src/main/java/com/nisum/kafka_poc/controller/KafkaConsumerController.java

+package com.nisum.kafka_poc.controller;
+
+import com.nisum.kafka_poc.service.KafkaConsumerService;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/kafka")
+public class KafkaConsumerController {
+
+    private final KafkaConsumerService kafkaConsumerService;
+
+    public KafkaConsumerController(KafkaConsumerService kafkaConsumerService) {
+        this.kafkaConsumerService = kafkaConsumerService;
+    }
+
+    // Trigger Kafka message consumption from the configured topic
+    @GetMapping("/consume")
+    public String consumeMessage() {
+        // This is an example to demonstrate the consumer; actual consumption happens automatically
+        return "Kafka Consumer is active, waiting for messages...";
+    }
+
+}

src/main/java/com/nisum/kafka_poc/controller/Test.java

+package com.nisum.kafka_poc.controller;
+
+import java.util.*;
+
+public class Test {
+
+    public static void main(String[] args) {
+        List<Integer> list = Arrays.asList(10,15,10,15,8,49,25,98,32);
+//        Set<Integer> data=new TreeSet<>(list);
+        list.stream().max(Integer::compareTo).get();
+    }
+}

src/main/java/com/nisum/kafka_poc/model/Employee.java

+package com.nisum.kafka_poc.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.springframework.web.bind.annotation.GetMapping;
+
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class Employee {
+
+    private Integer Id;
+    private String name;
+    private double salary;
+
+
+}

src/main/java/com/nisum/kafka_poc/model/Test.java

+package com.nisum.kafka_poc.model;
+
+public class Test {
+}

src/main/java/com/nisum/kafka_poc/practice/CountDownLatchExample.java

+package com.nisum.kafka_poc.practice;
+
+import java.util.concurrent.CountDownLatch;
+
+public class CountDownLatchExample implements Runnable {
+
+    private int id;
+    private CountDownLatch latch;
+
+    public CountDownLatchExample(int id, CountDownLatch latch) {
+        this.id = id;
+        this.latch = latch;
+    }
+
+    @Override
+    public void run() {
+        try {
+            latch.await(); // Wait for the signal to start
+            System.out.println("Thread " + id + " is running.");
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/java/com/nisum/kafka_poc/practice/CountDownLatchMain.java

+package com.nisum.kafka_poc.practice;
+
+import java.util.concurrent.CountDownLatch;
+
+public class CountDownLatchMain {
+
+    public static void main(String[] args) throws InterruptedException {
+        int numberOfThreads = 5;
+        CountDownLatch latch = new CountDownLatch(1); // Latch to synchronize start
+
+        for (int i = 1; i <= numberOfThreads; i++) {
+            new Thread(new CountDownLatchExample(i, latch)).start();
+        }
+
+        Thread.sleep(1000); // Ensure all threads are ready
+        System.out.println("Releasing latch... All threads start now!");
+        latch.countDown(); // Release all threads at the same time
+    }
+}

src/main/java/com/nisum/kafka_poc/practice/CyclicBarrierExample.java

+package com.nisum.kafka_poc.practice;
+
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+
+public class CyclicBarrierExample implements Runnable{
+
+    private final int id;
+    private final CyclicBarrier cyclicBarrier;
+
+    public CyclicBarrierExample(int id, CyclicBarrier cyclicBarrier) {
+        this.id = id;
+        this.cyclicBarrier = cyclicBarrier;
+    }
+
+
+    @Override
+    public void run() {
+
+        try {
+            System.out.println("Thread " + id + " is ready...");
+            cyclicBarrier.await(); // Wait for all threads to reach this point
+            System.out.println("Thread " + id + " is running at " + System.currentTimeMillis());
+        } catch (InterruptedException | BrokenBarrierException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}

src/main/java/com/nisum/kafka_poc/practice/CyclicBarrierMain.java

+package com.nisum.kafka_poc.practice;
+
+import java.util.concurrent.CyclicBarrier;
+
+public class CyclicBarrierMain {
+    public static void main(String[] args) {
+
+        int numberOfThreads = 5;
+       CyclicBarrier barrier = new CyclicBarrier(numberOfThreads,
+               () -> {
+                   System.out.println("All runners reached the checkpoint. Race continues!");
+               });
+        for (int i = 1; i <= numberOfThreads; i++) {
+            new Thread(new CyclicBarrierExample(i, barrier)).start();
+        }
+    }
+}

src/main/java/com/nisum/kafka_poc/practice/DistinctValues.java

+package com.nisum.kafka_poc.practice;
+
+import java.sql.Array;
+import java.util.*;
+
+public class DistinctValues {
+    public static void main(String[] args) {
+        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 2, 6, 3, 7, 8, 8, 9);
+        int arr[]=new int[]{1,4,5,6};
+        Set<Integer> set=new HashSet<>();
+        list.stream()
+                .sorted(Collections.reverseOrder()).forEach(System.out::println);
+                /*.filter(data -> set.add(data))
+                .forEach(System.out::println);*/
+//        int a=Arrays.stream(arr).min().getAsInt();
+//        System.out.println(a);
+
+    }
+
+}

src/main/java/com/nisum/kafka_poc/practice/FindDuplicate.java

+package com.nisum.kafka_poc.practice;
+
+import org.apache.kafka.shaded.com.google.protobuf.MapEntry;
+
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class FindDuplicate {
+    public static void main(String[] args) {
+        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 2, 6, 3, 7, 8, 8, 9);
+        Set<Integer> set=new HashSet<>();
+         list.stream()
+                        .filter(data -> !set.add(data))
+                                .forEach(System.out::println);
+
+    }
+}

src/main/java/com/nisum/kafka_poc/practice/ListToMap.java

+package com.nisum.kafka_poc.practice;
+
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
+
+public class ListToMap {
+    public static void main(String[] args) {
+        Mono<String> mono1 = Mono.just("Hello");
+        Mono<String> mono2 = Mono.error(new RuntimeException("Something went wrong"));
+
+        Mono<Tuple2<String, String>> result = mono1.zipWith(mono2);
+
+        result.subscribe(
+                tuple -> System.out.println("Combined: " + tuple.getT1() + " and " + tuple.getT2()),
+                error -> System.out.println("Error: " + error.getMessage())
+        );
+
+    }
+}

src/main/java/com/nisum/kafka_poc/practice/NonRepeatedCharacters.java

+package com.nisum.kafka_poc.practice;
+
+public class NonRepeatedCharacters {
+    public static void main(String[] args) {
+        String message = "Come here";
+        Character a =message.chars().mapToObj(c -> (char)c)
+                .filter(ch -> message.indexOf(ch)==message.lastIndexOf(ch))
+                .findFirst().get();
+        System.out.println(a);
+    }
+}

src/main/java/com/nisum/kafka_poc/practice/RepeatedCharacter.java

+package com.nisum.kafka_poc.practice;
+
+public class RepeatedCharacter {
+    public static void main(String[] args) {
+        String message = "Come here";
+        Character a =message.chars().mapToObj(c -> (char)c)
+                .filter(ch -> message.indexOf(ch)!=message.lastIndexOf(ch))
+                .findFirst().get();
+        System.out.println(a);
+    }
+}

src/main/java/com/nisum/kafka_poc/practice/StartsWithOne.java

+package com.nisum.kafka_poc.practice;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class StartsWithOne {
+    public static void main(String[] args) {
+        List<Integer> list = Arrays.asList(1, 2, 13, 4, 15, 2, 16, 3, 7, 8, 8, 9);
+        list.stream()
+                .map(s -> s.toString()).filter(data -> data.startsWith("1"))
+                .forEach(System.out::println);
+    }
+}

src/main/java/com/nisum/kafka_poc/service/KafkaConsumerService.java

+package com.nisum.kafka_poc.service;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+@Service
+public class KafkaConsumerService {
+
+    @KafkaListener(topics = "test", groupId = "test-group")
+    public void listen(String message) {
+        System.out.println("Received message: " + message);
+    }
+
+}

src/main/resources/application.properties

+spring.application.name=kafka-poc

src/main/resources/application.yml

+spring:
+  kafka:
+    consumer:
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    template:
+      default-topic: test
+
+server:
+  port: 9293
+

src/test/java/com/nisum/kafka_poc/KafkaPocApplicationTests.java

+package com.nisum.kafka_poc;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class KafkaPocApplicationTests {
+
+	@Test
+	void contextLoads() {
+	}
+
+}