diff --git a/build.gradle.kts b/build.gradle.kts
index 71596a4..6f44e62 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -12,6 +12,7 @@ repositories {
 dependencies {
     implementation("com.couchbase.client:java-client:3.5.1")
     implementation("org.apache.commons:commons-lang3:3.12.0")
+    implementation("org.apache.kafka:kafka-clients:3.6.1")
     implementation("org.apache.logging.log4j:log4j-core:2.14.1")
     implementation("org.apache.logging.log4j:log4j-slf4j-impl:2.14.1")
     implementation("redis.clients:jedis:3.7.0")
diff --git a/src/main/java/com/zingplay/tiennm5/KafkaMessageReceiver.java b/src/main/java/com/zingplay/tiennm5/KafkaMessageReceiver.java
new file mode 100644
index 0000000..65caa4b
--- /dev/null
+++ b/src/main/java/com/zingplay/tiennm5/KafkaMessageReceiver.java
@@ -0,0 +1,34 @@
+package com.zingplay.tiennm5;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+
+public class KafkaMessageReceiver {
+    public static void main(String[] args) {
+        // Set up the properties for the Kafka consumer
+        Properties props = new Properties();
+        props.put("bootstrap.servers", "localhost:9092"); // Replace with your Kafka server addresses
+        props.put("group.id", "test-group");
+        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+
+        // Create the consumer
+        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+
+        // Subscribe to the "test" topic
+        consumer.subscribe(Collections.singletonList("test"));
+
+        // Continuously listen for new messages
+        while (true) {
+            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
+            for (ConsumerRecord<String, String> record : records) {
+                System.out.printf("Received message: (%s, %s)%n", record.key(), record.value());
+            }
+        }
+    }
+}
diff --git a/src/main/java/com/zingplay/tiennm5/KafkaMessageSender.java b/src/main/java/com/zingplay/tiennm5/KafkaMessageSender.java
new file mode 100644
index 0000000..a41635b
--- /dev/null
+++ b/src/main/java/com/zingplay/tiennm5/KafkaMessageSender.java
@@ -0,0 +1,25 @@
+package com.zingplay.tiennm5;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import java.util.Properties;
+
+public class KafkaMessageSender {
+    public static void main(String[] args) {
+        // Set up the properties for the Kafka producer
+        Properties props = new Properties();
+        props.put("bootstrap.servers", "localhost:9092"); // Replace with your Kafka server addresses
+        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+
+        // Create the producer
+        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
+
+        // Send a message to the "test" topic
+        producer.send(new ProducerRecord<String, String>("test", "key", "value"));
+
+        // Close the producer
+        producer.close();
+    }
+}