Kafka for microservice intercommunication
I'm thrilled to guide you through the realm of Apache Kafka, a revolutionary solution that has transformed the way we build highly resilient and scalable micro-services. This technology has played a pivotal role in enabling seamless communication among microservices, underpinning a multitude of powerful applications.
For this project, we decided to use microservices architecture, where independent services need to talk to each other. That's where Kafka comes in as the communication tool.
Here are the key microservices in the ecosystem:
Central Service: This is the heart of the system, responsible for user management, handling payments, and managing access control lists (ACLs) to regulate user permissions for various apps.
Learning Management Service (LMS): A dedicated service for delivering and managing learning content.
Point of Sale (POS) Service: A specialized service for point-of-sale transactions.
Ticketing Service: Designed to manage and process ticket-related activities.
The central service acts as the linchpin, governing user access to different applications. It's in charge of maintaining critical user information and managing permissions for each application.
Now how does the central service speak with other services about which user has access to which application? That's where Kafka comes in!
Kafka steps in as the messenger. Here's how it works:
Publishing Access Data: When a user gains access to an app, the central service acts as a producer. It publishes a message containing all the necessary information about the user's new access rights.
Subscribing to Changes: On the other side, the target application subscribes to a specific topic. This allows it to consume messages related to access changes.
Choosing AWS MSK: For my project, I opted for Amazon Managed Streaming for Apache Kafka (MSK). This decision had a significant impact on streamlining my development process. With MSK, I could say goodbye to the complexities of managing brokers and ZooKeeper, freeing up more time to focus on the actual implementation.
In a nutshell, Apache Kafka, coupled with AWS MSK, proved to be the perfect fit for my microservices-based project. It offered the reliability, scalability, and real-time data streaming capabilities I needed to create a robust, interconnected ecosystem of services.
AWS Infrastructure Setup:
Before diving into the Kafka implementation, I set up the necessary AWS infrastructure. Here are the components I configured:
VPC (Virtual Private Cloud): I established a VPC with two public subnets across two availability zones (1a and 1b). This ensures high availability and fault tolerance for the Kafka cluster.
Security Groups: I created a security group that allows inbound traffic only from specific sources, such as my EC2 jumper instance and ECS services. This fine-grained control enhances the security of the Kafka cluster.
Kafka (AWS MSK): AWS MSK serves as the core Kafka infrastructure. It simplifies cluster management, making it easier to maintain.
Secrets Manager: For security, I used AWS Secrets Manager to store usernames and passwords required for SCRAM (Salted Challenge Response Authentication Mechanism) authentication. This ensures that only authorized entities can access the Kafka cluster.
IAM Role: An IAM role with full access to Kafka allows EC2 instances to communicate securely with the Kafka brokers. Role-based access control ensures that only authorized entities interact with the Kafka cluster.
EC2 Jumper Instance: I deployed an EC2 jumper instance, which plays a crucial role in managing administrative operations. This includes tasks like creating topics, configuring ACLs, and managing consumer groups. This separation of duties enhances security and control.
Kafka Brokers: I set up two Kafka brokers initially. Kafka's distributed nature allows easy scaling by adding more brokers as data volume and throughput requirements grow.
Topics and Partitions: I created Kafka topics (e.g., lms-users-acls, pos-users-acls, ticketing-users-acls) with multiple partitions (2 in each topic) to efficiently organize and distribute data.
Consumer Groups: To enable parallel message processing by multiple consumers, I established consumer groups (e.g., lms-users-acls-group, pos-users-acls-group, ticketing-users-acls-group). This is essential for achieving high throughput and scalability.
Users: I've differentiated user roles (e.g., central-user, lms-user, pos-user, ticketing-user) to implement access control and data segregation in line with microservices best practices.
User Permissions on Topics:
Central-user:
Permissions: Write access to topics - lms-users-acls, pos-users-acls, ticketing-users-acls.
Role: This user plays a central role and has write permissions for various topics, allowing it to publish messages.
Lms-user:
Permissions: Read access to the topic - lms-users-acls.
Role: This user is primarily concerned with learning management and has read-only access to the designated topic.
Pos-user:
Permissions: Read access to the topic - pos-users-acls.
Role: This user focuses on point-of-sale operations and is granted read-only access to the specific topic.
Ticketing-user:
Permissions: Read access to the topic - ticketing-users-acls.
Role: This user handles ticketing and customer support tasks, with read-only access to the dedicated topic.
ACLs on Consumer Groups for Reading Data from Topics:
Lms-users-acls-group:
Permissions: Read access to the topic - lms-users-acls.
Role: This consumer group is associated with the LMS microservice, allowing it to read messages from the designated topic.
Pos-users-acls-group:
Permissions: Read access to the topic - pos-users-acls.
Role: This consumer group corresponds to the point-of-sale service, enabling it to consume messages from its specific topic.
Ticketing-users-acls-group:
Permissions: Read access to the topic - ticketing-users-acls.
Role: This consumer group is linked to the ticketing service, enabling it to access and process messages from the assigned topic.
Authentication Mechanisms:
To ensure secure producer and consumer communication, I employed two authentication mechanisms:
SCRAM (Salted Challenge Response Authentication Mechanism):
Usage: SCRAM is used for authentication between producers and consumers. It ensures secure and authenticated communication within the Kafka ecosystem.
IAM (Identity and Access Management):
Usage: IAM is implemented for administrative privileges. It facilitates secure communication between the EC2 instances and Kafka brokers, granting administrative control only to authorized entities.
Cluster creation steps in AWS MSK
Review and create. Creation step takes around 15 to 20 minutes.
Once cluster is created, We need to attach scram users and IAM role in the security section.
Go to secrets manager and create users listed above with your chosen passwords.
Come back to cluster console and you would see the provision to attach these users in the console as shown. Select the users from secrets manager and attach. Attach IAM role as well
In my development environment setup, I need to connect to Kafka from my jumper instance, ECS services, and local machines for development and testing. Kafka offers both local and private endpoints for communication. For following this tutorial I am giving public access for the Kafka end point.
Please note that enabling public endpoints is not advisable for production environments and should be used solely for testing and development purposes.
Go to network section and click on edit public access, check the turn on and save changes.
After completing all these steps, it's time to explore the broker endpoints. Simply navigate to the Cluster Summary and click on 'View Cluster Configuration.' You'll find separate endpoints for both IAM and SCRAM authentications. When communicating with Kafka from an EC2 instance, we'll utilize the IAM endpoints, whereas for communication from ECS or local environments, the SCRAM endpoints will be employed.
Now, let's delve into a step-by-step guide on the essential commands for installing and configuring your Kafka cluster directly from an EC2 instance. These instructions encompass installing the Kafka command-line interface (CLI), configuring a client properties file to enable IAM authentication, and fundamental commands for various operations.
sudo yum -y install java-11
wget https://archive.apache.org/dist/kafka/{YOUR MSK VERSION}/kafka_2.13-{YOUR MSK KAFKA VERSION}.tgz
tar -xzf kafka_2.13-{YOUR MSK KAFKA VERSION}.tgz
Enable IAM auth
-> Install IAM handler in kafka_2.13-{YOUR MSK VERSION}/libs/ folder
wget https://github.com/aws/aws-msk-iam-auth/releases/download/v1.1.1/aws-msk-iam-auth-1.1.1-all.jar
-> create client.properties file at kafka_2.13-{YOUR MSK VERSION}/bin directory. Paste the below 5 lines in the file and save
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
-> Configure AWS CLI with the IAM role created crdentials. Make sure the IAM role has full permission to MSK. Follow the below command and provide the details obtained.
aws configure
Create Topics
-> kafka_2.13-{YOUR MSK VERSION}/bin/kafka-topics.sh --create --bootstrap-server {brokers-string} --command-config kafka_2.13-{YOUR MSK VERSION}/bin/client.properties --replication-factor 2 --partitions 2 --topic topicname
List and delete topics
-> List topics
kafka_2.13-{YOUR MSK VERSION}/bin/kafka-topics.sh --bootstrap-server {brokers-string} --list --command-config bin/client.properties
-> Delete topic
kafka_2.13-{YOUR MSK VERSION}/bin/kafka-topics.sh --delete --topic {topic name} --bootstrap-server {brokers-string} --command-config kafka_2.13-{YOUR MSK VERSION}/bin/client.properties
User ACls on Topics
-> Setup ACLS user. Say whether to give read/write permission to a specific topic.
kafka_2.13-{YOUR MSK VERSION}/bin/kafka-acls.sh --bootstrap-server {brokers-string} --add --allow-principal User:{username} --operation Write --topic topicname --command-config kafka_2.13-{YOUR MSK VERSION}/bin/client.properties
-> Remove Permission to topic on user
kafka_2.13-{YOUR MSK VERSION}/bin/kafka-acls.sh --bootstrap-server {brokers-string} --remove --allow-principal User:{username} --operation Write --topic {username} --command-config kafka_2.13-3.4.0/bin/client.properties
ACLS for consumer groups to read from topics
-> Add permission
kafka_2.13-{YOUR MSK VERSION}/bin/kafka-acls.sh --command-config kafka_2.13-{YOUR MSK VERSION}/bin/client.properties --bootstrap-server={brokers-string} --add --allow-principal User:{username} --operation {Read/Describe} --topic {topic name} --group {consumer group name}
-> Remove Permissison
kafka_2.13-{YOUR MSK VERSION}/bin/kafka-topics.sh --command-config kafka_2.13-{YOUR MSK VERSION}/bin/client.properties --bootstrap-server={brokers-string} --remove --allow-principal User:{username} --operation {Read/Describe} --topic {topic name} --group {consumer group name}
NOTE: SCRAM auth is username-password auth that we use in our development. Without ACLS setup, you will not be able to communicate to cluster using SCRAM details.
SCRAM users SCRAM-SHA-512 mechanism, SASL_SSL protocol.
We have completed the essential steps, and now we're prepared to configure our producers and consumers. Below, you'll find examples of my test producers and consumers for your reference:
Producer
from confluent_kafka import Producer
import json
# Kafka broker configuration
def produce_messages():
config = {
'bootstrap.servers': {KAFKA_BROKERS_STRING},
'sasl.mechanisms': 'SCRAM-SHA-512',
'security.protocol': 'SASL_SSL',
'sasl.username': "central-user",
'sasl.password': "test@123",
"client.id": "lms-users-acls-client"
}
# Create a Kafka producer instance
producer = Producer(config)
# Topic to send the message to
topic = 'lms-users-acls' # Replace with your topic name
# Message content
message_content = {
"message_uid": "123",
"role": "admin",
"action" : "add",
"user_uid":"123456",
"username":"test123",
"email": "test@gmail.com",
"first_name":"Chota",
"Last_name":"beam",
}
# Callback function to handle delivery reports
def delivery_report(err, msg):
if err is not None:
print('Message delivery failed: {}'.format(err))
else:
print('Message delivered to {} [{}]'.format(msg.topic(),msg.partition()))
# Produce the message
producer.produce(topic, value=json.dumps(
message_content).encode('utf-8'), callback=delivery_report)
# Wait for any outstanding messages to be delivered and delivery reports received
producer.flush()
produce_messages()
Consumer
def lms_credits_usage_messages_consumer_1():
try:
config = {
'bootstrap.servers': {KAFKA_BROKERS_STRING},
'sasl.mechanisms': 'SCRAM-SHA-512',
'security.protocol': 'SASL_SSL',
'sasl.username':'lams-user',
'sasl.password':'test@123',
'auto.offset.reset': 'earliest',
"group.id": "lms-users-acls-group",
"client.id": "lms-users-acls-consumer"
}
consumer = Consumer(config)
consumer.subscribe(["lms-users-acls"])
# run infinite until stopped
while True:
message = consumer.poll(1.0)
if not message:
continue
elif message.error():
continue
process_messages(message, "lms", db)
except Exception as e:
print(e)
Conclusion
As we wrap up this journey into the world of Apache Kafka and AWS MSK, we've explored the essential components, configurations, and steps required to establish a powerful and efficient communication framework for your microservices. By carefully crafting a Kafka cluster, setting up user permissions and ACLs, and implementing IAM and SCRAM authentication, you've laid the foundation for a robust and scalable microservices architecture. With these tools in your arsenal, your applications are poised to excel, delivering outstanding performance and user experiences. Thank you for joining me on this Kafka adventure, and may your microservices thrive and flourish.