DEV Community: Yusen Meng

Building a Multi-Agent Framework from Scratch with LlamaIndex

Yusen Meng — Tue, 29 Oct 2024 07:57:32 +0000

Tip: For a better reading experience, please visit the full article on GitHub.

Code: The complete source code for this tutorial is available in the GitHub repository.

Introduction

Multi-agent systems are like teams of specialized workers, each with their own expertise, working together to accomplish complex tasks. In this tutorial, we'll build a multi-agent system using LlamaIndex that generates high-quality Anki flashcards from any given text.

We'll start simple and gradually add more capabilities, similar to how you might build a team - starting with one person and gradually adding more specialists as needed.

Version 1: Basic Single-Agent System

In our first version, we'll create a simple system with just one agent that can generate Anki flashcards from text. Think of it as hiring your first employee who knows how to create educational content.

Architecture

graph LR
    Input[Input Text] -->|Process| QA[Q&A Generator]
    QA -->|Generate| Cards[Flashcards]

    style Input fill:#e1f5fe,stroke:#01579b
    style QA fill:#e8f5e9,stroke:#2e7d32
    style Cards fill:#fff3e0,stroke:#ef6c00

Core Components

Let's break down our code into manageable pieces:

Data Models: First, we define what our flashcards should look like using Pydantic:

class QACard(BaseModel):
    question: str
    answer: str
    extra: str

class Flashcard_model(BaseModel):
    cards: List[QACard]

LLM Setup: We configure LlamaIndex to use our preferred language model:

def get_shared_llm():
    """Returns a shared LLM instance for all agents."""
    return OpenAI(
        model="gpt-4o-mini", 
        temperature=0,
        api_base="http://127.0.0.1:4000/v1",
        api_key="sk-test"
    )

Agent Creation: Our Q&A Generator agent is specialized in creating flashcards:

def qa_generator_factory() -> OpenAIAgent:
    system_prompt = """
    You are an educational content creator specializing in Anki flashcard generation.
    Your task is to create clear, concise flashcards following these guidelines:

    1. Each card should focus on ONE specific concept
    2. Questions should be clear and unambiguous
    3. Answers should be concise but complete
    4. Include relevant extra information in the extra field
    5. Follow the minimum information principle

    Format each card as:
    <card>
        <question>Your question here</question>
        <answer>Your answer here</answer>
        <extra>Additional context, examples, or explanations</extra>
    </card>
    """

    return OpenAIAgent.from_tools(
        [],
        llm=get_shared_llm(),
        system_prompt=system_prompt,
    )

Response Handling: We use a transformer function to convert the agent's output into structured data:

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def transformer(message: str) -> dict:
    chat_prompt_tmpl = ChatPromptTemplate(
        message_templates=[
            ChatMessage.from_str(message, role="user")
        ]
    )
    llm = get_shared_llm()
    structured_data = llm.structured_predict(Flashcard_model, chat_prompt_tmpl)
    return structured_data.model_dump()

Main Function

Here's how we put it all together:

def generate_anki_cards(input_text: str) -> dict:
    # Create the agent
    agent = qa_generator_factory()

    # Generate flashcards
    response = agent.chat(
        f"Generate Anki flashcards from the following text:\n\n{input_text}"
    )

    # Transform the response into structured data
    structured_response = transformer(str(response))
    return structured_response

Example Usage

Let's try it with a sample text about RSI (Relative Strength Index):

sample_text = """
The Relative Strength Index (RSI) is a momentum indicator used in technical analysis.
It measures the speed and magnitude of recent price changes to evaluate overbought
or oversold conditions. RSI is displayed as an oscillator on a scale from 0 to 100.
Readings above 70 generally indicate overbought conditions, while readings below 30
indicate oversold conditions.
"""

flashcards = generate_anki_cards(sample_text)
print(flashcards)

Key Features

Single Responsibility: Our agent focuses solely on creating flashcards
Data Validation: Pydantic ensures our data is properly structured
Error Handling: Retry mechanism helps handle temporary failures
Structured Output: Consistent XML format for easy parsing
Type Safety: Python type hints help catch errors early

Current Limitations

This basic version has some limitations we'll address in future versions:

No quality control (no one reviewing the cards)
Limited understanding of complex topics
No way to handle specialized content (like code examples)
No memory of previous messages
No way to improve cards based on feedback

In the next version, we'll add a second agent to review and improve the generated flashcards, making our system more robust.

Version 2: Two-Agent System

In this version, we'll add a second agent - the Reviewer - to improve the quality of our flashcards. Think of it like having a teacher who creates the flashcards (Q&A Generator) and an editor who reviews and improves them (Reviewer).

Architecture

graph LR
    Input[Input Text] -->|Generate| QA[Q&A Generator]
    QA -->|Review| Rev[Reviewer]
    Rev -->|Output| Cards[Final Flashcards]

    style Input fill:#e1f5fe,stroke:#01579b
    style QA fill:#e8f5e9,stroke:#2e7d32
    style Rev fill:#f3e5f5,stroke:#7b1fa2
    style Cards fill:#fff3e0,stroke:#ef6c00

New Components

Let's look at what we're adding to our system:

Agent Types: First, we define our agent types using an Enum:

class Speaker(str, Enum):
    QA_GENERATOR = "Q&A Generator"
    REVIEWER = "Reviewer"

Reviewer Agent: This agent specializes in improving flashcard quality:

def reviewer_factory() -> OpenAIAgent:
    system_prompt = """
    You are the Reviewer agent. Your task is to review and refine Anki flashcards,
    ensuring they follow these rules:

    1. Each card should test ONE piece of information
    2. Questions must be:
       - Simple and direct
       - Testing a single fact
       - Using cloze format when appropriate
    3. Answers must be:
       - Brief and precise
       - Limited to essential information
    4. Extra field must include:
       - Detailed explanations
       - Examples
       - Context
    """

    return OpenAIAgent.from_tools(
        [],
        llm=get_shared_llm(),
        system_prompt=system_prompt,
    )

State Management: We track the progress of our flashcard creation:

def get_initial_state(text: str) -> dict:
    return {
        "input_text": text,
        "qa_cards": "",
        "review_status": "pending"
    }

Main Function

Here's how we combine both agents:

def generate_anki_cards(input_text: str) -> dict:
    # Initialize state
    state = get_initial_state(input_text)

    # Step 1: Generate initial cards
    generator = qa_generator_factory()
    response = generator.chat(
        f"Generate Anki flashcards from the following text:\n\n{state['input_text']}"
    )
    state["qa_cards"] = str(response)

    # Step 2: Review and improve cards
    reviewer = reviewer_factory()
    review_response = reviewer.chat(
        f"Review and improve these flashcards:\n\n{state['qa_cards']}"
    )
    state["qa_cards"] = str(review_response)

    # Transform final cards into structured data
    return transformer(state["qa_cards"])

Example Usage

Let's see how it works with our RSI example:

sample_text = """
The Relative Strength Index (RSI) is a momentum indicator used in technical analysis.
It measures the speed and magnitude of recent price changes to evaluate overbought
or oversold conditions. RSI is displayed as an oscillator on a scale from 0 to 100.
Readings above 70 generally indicate overbought conditions, while readings below 30
indicate oversold conditions.
"""

flashcards = generate_anki_cards(sample_text)

Example Output

Here's what the process looks like:

Initial Generation:

<cards>
    <card>
        <question>What is RSI in technical analysis?</question>
        <answer>A momentum indicator that measures speed and magnitude of recent price changes</answer>
        <extra>Used to evaluate overbought/oversold conditions</extra>
    </card>
</cards>

After Review:

<cards>
    <card>
        <question>RSI (Relative Strength Index) is a _____ indicator in technical analysis.</question>
        <answer>momentum</answer>
        <extra>
        Key points about RSI:
        - Measures speed/magnitude of price changes
        - Scale: 0 to 100
        - Overbought: Above 70
        - Oversold: Below 30
        </extra>
    </card>
</cards>

Key Improvements

Quality Control: The Reviewer agent ensures better card quality
Better Format: Cloze deletions for more effective learning
Richer Context: More detailed extra field content
Progress Tracking: State management shows card status
Consistent Structure: XML format maintained throughout

Current Limitations

While better than Version 1, this system still has some limitations:

No way to analyze complex topics
Basic two-step process (could be more sophisticated)
No memory of previous messages
Limited error handling
No way to handle specialized content (like code)

In Version 3, we'll add more agents to handle these limitations and create a more robust system.

Version 3: Multi-Agent Orchestra

In this version, we enhance our system by enabling multiple agents to collaborate effectively. Each agent has a specific role, and they work together to create high-quality flashcards. The Orchestrator agent plays a crucial role in managing this collaboration, ensuring that each agent contributes at the right time. Additionally, the Topic Analyzer helps break down complex topics into simpler parts, providing a structured overview of the content.

How Multi-Agent Collaboration Works

In a multi-agent system, each agent is like a team member with a specific expertise. Here's how they collaborate:

Specialized Roles: Each agent is designed to perform a specific task. For example, the Q&A Generator creates flashcards, while the Reviewer ensures their quality.
Sequential Processing: Agents work in a sequence, with each one building on the output of the previous agent. This ensures that the final product is refined and comprehensive.
Dynamic Decision-Making: The Orchestrator agent decides which agent should work next based on the current state of the process. This allows the system to adapt to the needs of the content dynamically.
Shared Context: A memory system maintains context across interactions, allowing agents to make informed decisions based on previous outputs.

Introducing the Orchestrator Agent

The Orchestrator is the manager of our multi-agent system. It coordinates the workflow, ensuring that each agent contributes effectively. Here's how it works:

Orchestrator's Role

Decision-Making: The Orchestrator evaluates the current state and decides which agent should run next. It ensures that the workflow is efficient and that all necessary tasks are completed.
Flexibility: The Orchestrator can choose any agent at any time, allowing for a flexible and adaptive process. It can run agents multiple times if needed or skip steps if the content is already complete.
End Condition: The Orchestrator decides when the process is complete, ensuring that the final flashcards are comprehensive and high-quality.

Orchestrator Implementation

Here's how we implement the Orchestrator:

def orchestrator_factory(state: dict) -> OpenAIAgent:
    system_prompt = f"""
    You are the Orchestrator agent. Your task is to coordinate the interaction between all agents to create high-quality flashcards.

    Current State:
    {pformat(state, indent=2)}

    Available agents:
    * Topic Analyzer - Breaks down complex topics
    * Q&A Generator - Creates flashcards
    * Reviewer - Improves card quality

    Decision Guidelines:
    - Use Topic Analyzer first for topic breakdown
    - Use Q&A Generator to create cards
    - Use Reviewer to refine cards
    - Choose END when the cards are ready

    Output only the next agent to run ("Topic Analyzer", "Q&A Generator", "Reviewer", or "END")
    """
    return OpenAIAgent.from_tools(
        [],
        llm=get_shared_llm(),
        system_prompt=system_prompt,
    )

Introducing the Topic Analyzer Agent

The Topic Analyzer is responsible for breaking down complex topics into manageable parts. It provides a structured overview of the content, which helps in creating more focused and comprehensive flashcards.

Topic Analyzer's Role

Content Breakdown: The Topic Analyzer identifies key topics and subtopics from the input text, helping to structure the content.
Hierarchical Structuring: It creates a hierarchical structure of topics, making it easier to understand the relationships between different concepts.
Application Suggestions: The agent suggests real-world applications or case studies related to the topics, enriching the flashcards with practical insights.

Topic Analyzer Implementation

Here's how we implement the Topic Analyzer:

def topic_analyzer_factory(state: dict) -> OpenAIAgent:
    system_prompt = f"""
    You are the Topic Analyzer agent. Your task is to analyze the given text and identify key topics for flashcard creation.

    Current State:
    {pformat(state, indent=2)}

    Instructions:
    1. Identify main concepts and sub-concepts
    2. Create a structured list of topics
    3. Suggest real-world applications

    Output format:
    <topics>
        <topic>
            <name>Main topic name</name>
            <subtopics>
                <subtopic>Subtopic 1</subtopic>
                <subtopic>Subtopic 2</subtopic>
            </subtopics>
        </topic>
    </topics>
    """
    return OpenAIAgent.from_tools(
        [],
        llm=get_shared_llm(),
        system_prompt=system_prompt,
        verbose=True
    )

Main Function

The main function now includes the Orchestrator, which dynamically decides the workflow:

def generate_anki_cards(input_text: str) -> dict:
    # Initialize state and memory
    state = get_initial_state(input_text)
    memory = setup_memory()

    while True:
        # Get current chat history
        current_history = memory.get()

        # Let Orchestrator decide next step
        orchestrator = orchestrator_factory(state)
        next_agent = str(orchestrator.chat(
            "Decide which agent to run next based on the current state.",
            chat_history=current_history
        )).strip().strip('"').strip("'")
        print(f"\nOrchestrator selected: {next_agent}")

        if next_agent == "END":
            print("\nOrchestrator decided to end the process")
            break

        # Execute selected agent
        try:
            if next_agent == Speaker.TOPIC_ANALYZER.value:
                analyzer = topic_analyzer_factory(state)
                response = analyzer.chat(
                    f"Analyze this text for flashcard topics:\n\n{state['input_text']}",
                    chat_history=current_history
                )
                state["topics"] = str(response)
                print("\nTopic Analysis Results:")
                print(state["topics"])

            elif next_agent == Speaker.QA_GENERATOR.value:
                generator = qa_generator_factory()
                response = generator.chat(
                    f"Generate flashcards for this topic:\n\n{state['topics']}",
                    chat_history=current_history
                )
                state["qa_cards"] = str(response)
                state["review_status"] = "needs_review"
                print("\nGenerated Cards:")
                print(state["qa_cards"])

            elif next_agent == Speaker.REVIEWER.value:
                reviewer = reviewer_factory()
                response = reviewer.chat(
                    f"Review these flashcards:\n\n{state['qa_cards']}",
                    chat_history=current_history
                )
                state["qa_cards"] = str(response)
                state["review_status"] = "reviewed"
                print("\nReviewed Cards:")
                print(state["qa_cards"])

                # Update memory with new interaction
                memory.put(ChatMessage(role="assistant", content=str(response)))
                print(f"\nUpdated memory with {next_agent}'s response")

        except Exception as e:
            print(f"\nError in {next_agent}: {str(e)}")
            continue

    # Transform final cards into structured data
    final_cards = transformer(state["qa_cards"])
    return final_cards

Example Usage

Here's how the orchestrated system works with our RSI example:

if __name__ == "__main__":
    sample_text = """
    The Relative Strength Index (RSI) is a momentum indicator used in technical analysis.
    It measures the speed and magnitude of recent price changes to evaluate overbought
    or oversold conditions. RSI is displayed as an oscillator on a scale from 0 to 100.
    Readings above 70 generally indicate overbought conditions, while readings below 30
    indicate oversold conditions. The RSI can also help identify divergences, trend lines,
    and failure swings that may not be apparent on the underlying price chart.
    """

    flashcards = generate_anki_cards(sample_text)
    print("\nGenerated, Analyzed, and Reviewed Flashcards:")
    print(flashcards)

Key Improvements

Coordinated Workflow: The Orchestrator manages agent interactions, ensuring a smooth and efficient process.
Topic Analysis: The Topic Analyzer provides a deeper understanding of content structure, leading to more comprehensive flashcards.
Persistent Memory: Context is maintained across interactions, allowing for more informed decision-making.
Flexible Execution: The Orchestrator dynamically selects agents based on the current state, adapting to the needs of the content.
Hierarchical Processing: Topics are broken down into manageable chunks, making it easier to create detailed and accurate flashcards.

Limitations

While this version is more sophisticated, it still has some areas for improvement:

No specialized handling of code examples
Limited formatting options
Basic error handling

In Version 4, we'll introduce the Code and Extra Field Expert agent and a Formatter agent to handle code examples and ensure consistent formatting. We'll also improve error handling and add support for different content types.

Version 4: Enhanced Capabilities

In this version, we introduce two new agents: the Code and Extra Field Expert and the Formatter. These agents handle code examples and ensure consistent formatting, respectively. We also improve error handling and add support for different content types.

Architecture

graph TB
    Input[Input Text] -->|Start| Orch[Orchestrator]

    Orch -->|Analyze| TA[Topic Analyzer]
    Orch -->|Process Code| Code[Code Expert]
    Orch -->|Generate| QA[Q&A Generator]
    Orch -->|Format| Format[Formatter]
    Orch -->|Review| Rev[Reviewer]

    TA -->|Topics| Decision{Continue?}
    Code -->|Results| Decision
    QA -->|Results| Decision
    Format -->|Results| Decision
    Rev -->|Results| Decision

    Decision -->|Yes| Orch
    Decision -->|No| Final[Final Flashcards]

    style Input fill:#e1f5fe,stroke:#01579b
    style Orch fill:#fff3e0,stroke:#ef6c00
    style TA fill:#e8f5e9,stroke:#2e7d32
    style Code fill:#e8f5e9,stroke:#2e7d32
    style QA fill:#bbdefb,stroke:#1976d2
    style Format fill:#bbdefb,stroke:#1976d2
    style Rev fill:#f3e5f5,stroke:#7b1fa2
    style Final fill:#fff3e0,stroke:#ef6c00
    style Decision fill:#ffecb3,stroke:#ffa000

New Components

1. Updated Agent Types

We expand our agent types to include the new roles:

class Speaker(str, Enum):
    QA_GENERATOR = "Q&A Generator"
    REVIEWER = "Reviewer"
    TOPIC_ANALYZER = "Topic Analyzer"
    ORCHESTRATOR = "Orchestrator"
    CODE_AND_EXTRA_FIELD_EXPERT = "Code and Extra Field Expert"
    FORMATTER = "Formatter"

2. Code and Extra Field Expert

This agent enhances flashcards with code examples and detailed explanations:

def code_and_extra_field_expert_factory() -> OpenAIAgent:
    system_prompt = """
    You are the Code and Extra Field Expert agent. Your task is to enhance Anki flashcards
    by adding relevant code snippets and comprehensive extra content.

    Instructions:
    1. Add clear, concise code examples that illustrate key concepts
    2. Ensure code snippets are well-commented and easy to understand
    3. In the extra field, provide:
       - Step-by-step explanations of code snippets 
       - Common use cases and scenarios
       - Potential pitfalls and edge cases
       - Best practices and optimization tips
    4. Use appropriate markdown formatting for code blocks
    5. Include relevant documentation links
    6. Ensure explanations are clear for a 15-year-old
    """
    return OpenAIAgent.from_tools(
        [],
        llm=get_shared_llm(),
        system_prompt=system_prompt,
    )

3. Formatter Agent

This agent ensures that the flashcards are properly formatted:

def formatter_agent_factory() -> OpenAIAgent:
    system_prompt = """
    You are the Formatter agent. Your task is to ensure proper XML structure and markdown
    formatting in the flashcards.

    Formatting Rules:
    1. Maintain valid XML structure
    2. Properly escape special characters
    3. Format code blocks with appropriate language tags
    4. Use consistent indentation
    5. Ensure markdown compatibility
    6. Preserve code snippets exactly as provided
    7. Handle nested structures properly
    """
    return OpenAIAgent.from_tools(
        [],
        llm=get_shared_llm(),
        system_prompt=system_prompt,
    )

Enhanced Error Handling and Validation

We improve error handling with retries and validation:

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def validate_and_transform(message: str) -> dict:
    try:
        # Transform to structured data
        chat_prompt_tmpl = ChatPromptTemplate(
            message_templates=[
                ChatMessage.from_str(message, role="user")
            ]
        )
        structured_data = get_shared_llm().structured_predict(
            Flashcard_model, 
            chat_prompt_tmpl
        )
        return structured_data.model_dump()

    except ET.ParseError as e:
        print(f"XML validation error: {str(e)}")
        raise TryAgain
    except Exception as e:
        print(f"Transformation error: {str(e)}")
        raise TryAgain

Example Usage

Here's how the enhanced system works with a sample text that includes code:

if __name__ == "__main__":
    sample_text = """
    To calculate the RSI (Relative Strength Index) in Python, you typically use 
    technical analysis libraries like pandas-ta or the ta library. The RSI is 
    calculated using the average gains and losses over a specified period 
    (usually 14 periods). Here's how you can implement it:

    1. Using the ta library:
    ```

python
    import pandas as pd
    import ta

    # Assuming you have price data in a DataFrame
    df['RSI'] = ta.momentum.RSIIndicator(
        close=df['close'],
        window=14
    ).rsi()


    ```

    2. Manual implementation:
    ```

python
    def calculate_rsi(data, periods=14):
        delta = data.diff()
        gain = (delta.where(delta > 0, 0)).rolling(
            window=periods).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(
            window=periods).mean()
        rs = gain / loss
        return 100 - (100 / (1 + rs))


    ```
    """

    flashcards = generate_anki_cards(sample_text)
    print("Generated Flashcards with Code Examples:")
    print(flashcards)

Key Improvements

Code Handling: Specialized agent for code examples and explanations
Consistent Formatting: Dedicated formatter agent
Robust Error Handling: Improved validation and retries
Rich Extra Content: Comprehensive explanations and best practices
Markdown Support: Proper formatting of code blocks and text

Conclusion

Throughout this tutorial, we've built a sophisticated multi-agent system for generating Anki flashcards, progressing from a simple single-agent implementation to a complex orchestrated system. Let's recap the key developments across each version:

Evolution of the System

Version 1: Basic Single-Agent
- Established core data structures
- Implemented basic flashcard generation
- Set up error handling foundation
Version 2: Two-Agent Interaction
- Added quality control through review process
- Introduced state management
- Implemented iterative refinement
Version 3: Multi-Agent Orchestra
- Added orchestration layer
- Implemented topic analysis
- Enhanced context management through memory system
Version 4: Enhanced Capabilities
- Added specialized code handling
- Implemented consistent formatting
- Enhanced error handling and validation

Key Takeaways

Modular Design
- Each agent has a specific responsibility
- Easy to add new agents or modify existing ones
- Clear separation of concerns
Robust Architecture
- State management across agents
- Error handling and retry mechanisms
- Memory system for context preservation
Quality Control
- Multiple layers of review and refinement
- Consistent formatting and structure
- Rich content with code examples and explanations

Future Enhancements

While our system is already quite capable, there are several potential areas for future improvement:

Parallel Processing
- Implement concurrent card generation
- Add batch processing capabilities
- Optimize for large-scale operations
Advanced Features
- Image handling and processing
- Multi-language support
- Audio content integration
- Interactive examples
Learning Capabilities
- Feedback incorporation
- Quality metrics tracking
- Adaptive prompt refinement
Integration Options
- API endpoints
- Web interface
- Direct Anki integration
- Version control for cards

Final Thoughts

Building a multi-agent system requires careful consideration of agent interactions, state management, and error handling. This tutorial has demonstrated how to incrementally build such a system, starting from basic components and progressively adding sophistication.

The final system showcases the power of combining multiple specialized agents, each with its own expertise, to create a robust and flexible solution. While the focus has been on Anki flashcard generation, the patterns and principles demonstrated here can be applied to many other multi-agent applications.

Remember that the key to successful multi-agent systems lies in:

Clear agent responsibilities
Robust communication patterns
Comprehensive error handling
Flexible state management
Consistent output formatting

By following these principles and the patterns demonstrated in this tutorial, you can build your own multi-agent systems for various applications beyond flashcard generation.

Understanding and Mitigating Message Loss in Apache Kafka

Yusen Meng — Thu, 25 Apr 2024 07:14:36 +0000

Apache Kafka, a popular distributed streaming platform, enables the building of scalable and fault-tolerant real-time data pipelines and applications. However, like any distributed system, Kafka is not immune to message loss, which can lead to data inconsistencies and impact system reliability.
In this article, we will explore the potential causes of message loss in Apache Kafka, focusing on producers, brokers, and consumers. We'll discuss scenarios such as improper acknowledgment settings, asynchronous disk flushing, replica synchronization issues, and the pitfalls of auto-commit in consumers.
By understanding these message loss scenarios, we can develop strategies to mitigate risks and ensure the reliability of our Kafka-based applications. We'll cover techniques like proper configuration, manual offset committing, idempotent message processing, and transactional commits.
So, let's dive in and learn how to build robust and fault-tolerant Kafka-based systems that minimize message loss and guarantee data integrity.

Producer

When we call producer.send() to send a message, it doesn't get sent to the broker directly. There are two threads and a queue involved in the message-sending process:

Application thread
Record accumulator
Sender thread (I/O thread)

The flow of a message from the producer to the broker can be visualized as follows:

Application Thread:

The application thread is responsible for creating the producer record and calling the producer.send() method.
It serializes the key and value of the record using the configured serializers.
The record is then added to the record accumulator.

Record Accumulator:

The record accumulator is a buffer that holds the records waiting to be sent to the Kafka broker.
It is essentially a queue that stores the records in memory.
The records are grouped into batches based on the topic and partition they belong to.
The batch size and the time to wait before sending the batch can be configured using the 'batch.size' and 'linger.ms' properties, respectively.

Sender Thread:
- The sender thread, also known as the I/O thread, is responsible for sending the batches of records from the record accumulator to the Kafka broker.
- It periodically checks the record accumulator for any ready batches and sends them to the corresponding broker.
- The sender thread handles the network communication with the Kafka broker.

To ensure that messages are reliably sent to the broker and to prevent message loss, we need to configure the following properties:

'acks': This property determines the level of acknowledgment required from the broker.
- 'acks=0': The producer does not wait for any acknowledgment from the broker. Messages may be lost if the broker fails.
- 'acks=1' (default): The producer waits for the leader to acknowledge the write. Messages may be lost if the leader fails before replicating to followers.
- 'acks=all' or 'acks=-1': The producer waits for the full set of in-sync replicas to acknowledge the write. This provides the highest level of durability but increases latency.
'retries': This property specifies the number of times the producer will attempt to resend a failed record.
- By default, 'retries' is set to 0, meaning no retries will be attempted.
- Setting 'retries' to a higher value, such as 3 or 5, allows the producer to handle transient failures and improve the chances of successful message delivery.

Example:



import (
    "github.com/Shopify/sarama"
)

config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Retry.Max = 3

producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
if err != nil {
    // Handle error
}
defer producer.Close()

msg := &sarama.ProducerMessage{
    Topic: "my-topic",
    Key:   sarama.StringEncoder("key"),
    Value: sarama.StringEncoder("value"),
}

partition, offset, err := producer.SendMessage(msg)
if err != nil {
    // Handle error
}

Broker

A broker cluster should not lose messages when it is functioning normally. However, we need to understand which extreme situations might lead to message loss:

Asynchronous Disk Flush:

Kafka brokers use an asynchronous disk flush mechanism to improve I/O throughput and performance.
When a producer sends a message to a broker, the message is initially stored in memory and not immediately written to disk.
The broker periodically flushes the messages from memory to disk in the background.
If the broker instance crashes or experiences a sudden shutdown before the flush occurs, the messages that were in memory but not yet flushed to disk will be lost.
To mitigate this risk, Kafka provides configurable flush settings, such as 'log.flush.interval.messages' and 'log.flush.interval.ms', which determine when the broker should flush messages to disk.

Replica Synchronization:
- Kafka ensures data durability and fault tolerance by maintaining replicas of partitions across multiple broker instances.
- Each partition has a leader replica and zero or more follower replicas.
- The leader replica handles all read and write operations for the partition, while the follower replicas passively replicate the data from the leader.
- If the leader replica fails, one of the follower replicas is elected as the new leader to ensure data availability.
- However, if the replicas are not properly synchronized or if there is a significant lag between the leader and followers, message loss can occur in certain scenarios:
  - If the leader replica fails and a follower replica with outdated data is elected as the new leader, any messages that were not yet replicated to the follower will be lost.
  - If a producer sends a message with 'acks=1' (only wait for the leader acknowledgment) and the leader fails before replicating the message to followers, the message will be lost.
- To minimize the risk of message loss due to replica synchronization issues, it is important to configure the replication factor and the 'min.insync.replicas' setting appropriately.
- The replication factor determines the number of replicas for each partition, providing redundancy.
- The 'min.insync.replicas' setting specifies the minimum number of in-sync replicas required for a write to be considered successful, ensuring data consistency.

Here's a diagram illustrating the replication process in Kafka:

In this diagram, the producer sends a message to the leader replica, which then replicates the message to the follower replica. Once the follower replica acknowledges the replication, the leader replica sends an acknowledgment back to the producer.

To ensure data durability and minimize message loss, it is recommended to:

Configure a sufficient replication factor (e.g., 3) to maintain multiple copies of the data.
Set 'min.insync.replicas' to a value greater than 1 to ensure that writes are considered successful only when the message is replicated to multiple replicas.
Use 'acks=all' or 'acks=-1' to wait for acknowledgment from all in-sync replicas before considering a write successful.

By following these best practices and configuring Kafka appropriately, the risk of message loss in a broker cluster can be significantly reduced, even in extreme situations.

Consumer

Kafka offers different ways to commit offsets, which represent the position of the last consumed message in a partition. Offset committing is crucial for tracking the progress of a consumer and ensuring that messages are not processed multiple times or skipped. However, the auto-commit feature in Kafka can sometimes lead to message loss if not used carefully.

Auto-Commit:

By default, Kafka consumers have auto-commit enabled, which means the consumer will automatically commit the offsets of the messages it has received at a regular interval.
The auto-commit interval is controlled by the 'auto.commit.interval.ms' configuration property, which defaults to 5 seconds.
When auto-commit is enabled, the consumer sends an asynchronous commit request to the Kafka broker at the specified interval, acknowledging the messages it has received and processed.
However, auto-commit can be problematic in certain scenarios, particularly when the consumer is down or fails in the middle of processing a batch of messages.

Potential Message Loss with Auto-Commit:

Consumer Failure during Message Processing:

Consider a scenario where the consumer has received a batch of messages and auto-commit is enabled.
The consumer starts processing the messages but encounters an error or crashes before completing the processing of all messages in the batch.
Since the offsets were automatically committed at the beginning of the batch, the consumer will resume from the next offset when it restarts, even though some messages in the previous batch were not fully processed.
This can lead to message loss because the unprocessed messages will be skipped, and the consumer will move on to the next batch.

Delayed Message Processing:
- In some cases, message processing might take longer than the auto-commit interval.
- If the consumer is still processing a message when the auto-commit interval elapses, the offset will be committed before the message processing is complete.
- If the consumer fails or is shut down after the auto-commit but before completing the message processing, the message will be considered processed even though it was not fully handled.

To mitigate the risk of message loss with auto-commit, consider the following approaches:

Manual Offset Committing:

Instead of relying on auto-commit, you can manually control when offsets are committed.
After successfully processing a message or a batch of messages, the consumer can call the commitSync() or commitAsync() method to commit the offsets explicitly.
This ensures that offsets are only committed when the messages are fully processed, reducing the chances of message loss.

Idempotent Message Processing:

Design your consumer to handle message processing in an idempotent manner.
Idempotence means that the processing of the same message multiple times should have the same effect as processing it once.
By making the message processing idempotent, even if a message is processed more than once due to offset commit issues, it will not have unintended consequences.

Batch Processing with Transactional Commits:
- If your consumer processes messages in batches, you can use transactional commits to ensure atomic offset commits.
- With transactional commits, the offset commit is tied to the successful processing of the entire batch.
- If the batch processing fails, the transaction is aborted, and the offsets are not committed, allowing the consumer to reprocess the batch from the beginning.

Here's an example of manual offset committing in using the Kafka consumer API:



import (
    "context"
    "fmt"
    "github.com/Shopify/sarama"
)

func main() {
    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true

    // Create a new consumer
    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
    if err != nil {
        panic(err)
    }
    defer func() {
        if err := consumer.Close(); err != nil {
            panic(err)
        }
    }()

    // Subscribe to the topic
    topic := "my-topic"
    partitionConsumer, err := consumer.ConsumePartition(topic, 0, sarama.OffsetNewest)
    if err != nil {
        panic(err)
    }
    defer func() {
        if err := partitionConsumer.Close(); err != nil {
            panic(err)
        }
    }()

    // Consume messages
    for {
        select {
        case msg := <-partitionConsumer.Messages():
            // Process the message
            fmt.Printf("Received message: %s\n", string(msg.Value))

            // Manually commit the offset after processing the message
            partitionConsumer.MarkOffset(msg.Offset+1, "")
        case err := <-partitionConsumer.Errors():
            fmt.Println("Error:", err)
        }
    }
}

Another condition is that the consumer may consume the message multiple times.

In the given scenario, the Kafka consumer is configured with auto-commit enabled, which means that the consumer automatically commits the offsets of the messages it has consumed at regular intervals specified by the auto-commit interval.

The consumer starts by polling a batch of messages (batch 1) from the Kafka broker. It then processes messages 1, 2, and 3 from batch 1. During the processing of batch 1, the auto-commit interval elapses, but the consumer does not explicitly commit the offsets for batch 1 to the Kafka broker.

Next, the consumer polls the next batch of messages (batch 2) from the Kafka broker and starts processing messages 4 and 5 from batch 2. However, while processing batch 2, the consumer fails or crashes.

When the consumer restarts, it resumes from the last committed offset, which is the offset before batch 1 (since no explicit offset commit was made). The consumer polls messages from the last committed offset, which includes messages from batch 1 and batch 2. As a result, the consumer reprocesses messages 4 and 5 from batch 2, along with messages 6 and 7.

The issue with this scenario is that messages 4 and 5 are reprocessed because the offsets were not explicitly committed to the Kafka broker before the consumer failure. Reprocessing messages can lead to challenges such as duplicate processing or inconsistent state.

Conclusion

In this article, we explored the potential causes of message loss in Kafka and discussed strategies to mitigate them. We covered message loss scenarios in Kafka producers, brokers, and consumers, highlighting the importance of proper configuration and handling.

For producers, configuring the appropriate acks and retries settings ensures reliable message delivery. In the case of brokers, asynchronous disk flushing and replica synchronization issues can lead to message loss, which can be mitigated by configuring flush settings, replication factors, and min.insync.replicas.

Consumers, on the other hand, need to be cautious when using auto-commit, as it can result in message loss if the consumer fails during message processing. To prevent this, manual offset committing, idempotent message processing, and transactional commits can be employed.

Additionally, we discussed the scenario where a consumer may consume the same message multiple times due to the lack of explicit offset commits before a consumer failure. This can lead to duplicate processing and inconsistent states. To address this, manual offset committing or transactional processing can be used to ensure that offsets are only committed when messages are fully processed.

By understanding the potential pitfalls and applying the appropriate strategies, developers can build reliable and fault-tolerant Kafka-based systems. It is crucial to carefully consider the configuration settings, offset commit strategies, and error handling mechanisms to ensure data integrity and prevent message loss.

As with any distributed system, testing and monitoring play a vital role in identifying and resolving issues related to message loss. Regular monitoring of Kafka metrics, logs, and consumer lag can help detect anomalies and take corrective actions promptly.

In conclusion, while Kafka provides a robust and scalable platform for real-time data streaming, it is important to be aware of the potential message loss scenarios and employ the necessary measures to mitigate them. By following best practices and implementing the appropriate strategies discussed in this article, developers can build reliable and resilient Kafka-based applications that ensure data integrity and minimize the risk of message loss.

From the function of API Gateway, see the evolution of Web architecture

Yusen Meng — Wed, 01 Nov 2023 14:59:36 +0000

This article summarizes the evolution of Web service architecture from start-ups to large enterprises based on my own past work experience. These experiences come from practices in different companies and different levels of project complexity. These architectures come from practical experience in work, may not be complete, and there may be omissions or inaccuracies. Welcome to discuss and exchange, so that we can learn and progress together.

Monolithic architecture, a handy Swiss army knife

A new project or a start-up company, the most common Web architecture is monolithic architecture. It can still be seen in many small teams in the company. As the name suggests, in this architecture, all functions and components are packaged into a single, tightly integrated program and run in a process space. Strictly speaking, there is no API Gateway service here, all requests will be forwarded to the same Server through NGINX or Apache's reverse proxy.

Monolithic architecture has the following characteristics:

Tightly integrated components: In a monolithic application, all functions and components are tightly integrated. They may interact through function calls, library references, and other mechanisms.
Simple deployment: Since it is a monolithic application or service, deployment is usually relatively simple. Just need to start an application or service, in the early days when virtualization technology has not yet appeared, simple deployment and startup reduce the complexity of the project.
Scalability issues: As the application scale and complexity grow, it is easy to produce monolithic applications, and the application becomes difficult to maintain and expand. Especially when horizontal expansion is needed, we can only deploy the entire application to multiple servers, rather than deploying the bottleneck part.
Fault isolation: Since all functions are in one application, a component failure may cause the entire application to crash, affecting the integrity of the entire application.
Technology stack restrictions: In a monolithic application, usually use a technology stack or framework. If you need to introduce a new technology stack or framework, you often need to do a complete reconstruction.
Continuous integration/continuous deployment: As the complexity and volume of the application increase, building, testing, and deployment may become slower.

Monolithic architecture has been very popular in early Internet companies and has long dominated the mainstream. Therefore, we often hear about deploying a service to a 48C, 48G server. However, with the rise of containerization technology, this kind of monolithic application gradually disappeared in the long river of history. We now hear more about deploying a service to a 2C, 2G server, and then supporting more traffic and services through horizontal expansion.

The embryonic form of microservices, splitting monolithic applications into multiple small services

In response to the various problems encountered by monolithic applications, we have split the monolithic applications according to functions, which is usually judged by the independence of business logic or data. For example, user management, order processing, inventory management, etc. can be regarded as independent services. Next, we need to set up a service gateway, such as NGINX, to handle all requests and responses. The main responsibility of the service gateway is to forward the request to the corresponding service according to the path of the request. For example, all requests starting with /user are forwarded to the user management service, and all requests starting with /order are forwarded to the order processing service.

This method splits the original monolithic application, isolates according to different functions, reduces the complexity and maintenance cost of the application, increases the scalability and robustness of the application, and to a certain extent solves the problems brought by the previous monolithic application.

But this Web architecture also has some problems of its own.

Direct communication between the client and the microservice, strong coupling between the Server and the client. The subsequent iteration and upgrade of the Server are difficult, and the compatibility of old applications needs to be considered.
Part of an application often involves multiple services, requires multiple requests, and then aggregates data at the client level. The workload of the client increases, and the delay may increase due to the increase in the number of requests. Generally speaking, if our application wants to iterate quickly, it needs to be light in the front and heavy in the back. The front end mainly handles UI display logic, and the iteration speed is fast. If there is a lot of front-end logic, it will affect the delivery speed of the business.
It is not conducive to the unification of protocols. Because of the direct communication between the client and the server, the server needs to adapt to multiple ends. A function needs to provide different APIs for the client and the web end, and the complexity of API docking increases.
The compatibility of the client is coupled to the internal service, making the server need to handle complex compatibility logic.
Unified logic cannot converge, such as authentication, flow control, circuit breaking, etc., it is very difficult to handle scattered in various Servers.

BFF, the appearance of the savior

In order to solve the above problems, we follow the first principle of Internet engineering, that is, if the problem cannot be solved, add a layer on it. Therefore, we added an isolation layer in the middle:

By adding a layer of service directly facing the customer, the Server and the client are isolated. We do data aggregation and processing in this layer of service, provide APIs and data for front-end business scenarios. The underlying microservices use the same protocol for data export, only need to provide coarse-grained APIs. Facing the front end and the client, Node.js has a natural advantage, so the Node.js ecosystem has also become popular with this business architecture. This layer of service is the BFF (Backend for frontend) layer we are familiar with today.

There are many advantages after joining BFF:

Lightweight interaction: Each service only needs to consider the intranet interface and provide coarse-grained APIs. The protocol has been simplified.
Decoupling and rapid iteration: The work of data trimming and aggregation is placed in the BFF layer, and the API is customized for end users. The underlying microservices do not need to consider the end-user scenario, can iterate quickly and upgrade dynamically, only need to ensure the compatibility of the original interface and system.
The improvement of communication efficiency, the cooperation mode evolves into Web Server + Core Server.

BFF split, increase the robustness of the application

Using a single BFF layer as the only exit has an unavoidable problem, which is easy to cause single point failure. To solve this problem, we split BFF according to business scenarios, split into different BFFs, and are responsible by different teams, such as Order BFF, User BFF, Item BFF.

Many horizontal functional logics, such as security authentication, log monitoring, flow control, circuit breaking, etc., can be aggregated into a new gateway layer. Here we introduce a new API Gateway gateway layer, which sinks all common horizontal functions such as routing, authentication, flow control, and security.

API Gateway plays a crucial role, it realizes the decoupling and separation of a single BFF, so that each business line team can independently develop and deliver their own microservices. The common horizontal logic has been transferred from BFF to the gateway, so that BFF developers can focus more on the delivery of business logic, achieving separation of concerns (Separation of Concerns).

The actual business flow has also evolved into:

Client/Browser -> NLB -> API Gateway -> BFF -> Microservice

According to Conway's law, organizational structure will affect software architecture, and software architecture reflects organizational structure. At this stage, the engineering team is usually split into: Gateway Team, BFF Web Server Team, and Core Server Team.

Summary

There is no silver bullet in this world, the appearance and existence of each architecture has its specific advantages and disadvantages. Blindly choosing the most complex architecture is not always the best choice. According to the complexity and scenario of the business, as well as the human resources and engineering capabilities of the team, choosing the most suitable architecture is the optimal solution. As the business grows, we should gradually evolve the technical architecture and choose the architecture that is most suitable for the current situation.

Understanding Two-Phase and Three-Phase Commit Protocols in Distributed Systems

Yusen Meng — Mon, 30 Oct 2023 13:52:45 +0000

Introduction: What problems might we encounter in a distributed system?

In a distributed system, multiple nodes need to work together to complete various tasks. Take an e-commerce website as an example, the services of the website are distributed across multiple data centers. Users place orders on the website, the order service is in one data center, and the inventory service is in another data center. Without a coordination mechanism (such as two-phase commit), the following problems may occur: the order service has already created the order, but when reducing the inventory, due to network problems, the inventory service did not receive the request, resulting in incorrect inventory quantity, and users may purchase goods that are actually sold out. Or, the inventory service has already reduced the inventory, but there was a problem when the order service created the order, causing the inventory to be reduced erroneously and there is no corresponding order.

This is just one example of many possibilities. Without a suitable distributed transaction processing mechanism, it may lead to data inconsistency, system unreliability, and damage to the user's shopping experience. To solve these problems, we need to introduce protocols like two-phase commit to ensure that transactions in distributed systems can be executed safely and completely.

Two-Phase Commit: A Deep Understanding of the Core of Distributed Transactions

With the popularity of distributed systems, how to ensure data consistency across multiple nodes has become an important challenge. Two-phase commit (2PC) is one of the classic algorithms to solve the distributed transaction problem. This article will provide an in-depth analysis of the working principle of two-phase commit and its advantages and disadvantages.

What is Two-Phase Commit?

Two-phase commit (2PC) is a protocol that implements transaction atomicity in a distributed system. Atomicity means that a transaction (transaction) is either fully executed or not executed at all, and there will be no partial execution. In a distributed system, a transaction may involve multiple nodes, so a mechanism is needed to ensure that all nodes either commit (commit) the transaction or roll back (rollback) the transaction. This is the role of the two-phase commit protocol.

Basic Process of Two-Phase Commit

2PC mainly involves two roles: Coordinator and Participant.

First Phase (Preparation Phase)

Voting request: The coordinator sends a transaction request to all participants, asking them whether they can commit the transaction.
Voting: Each participant pre-executes the transaction and decides to vote based on its local situation. If it can be executed, it records the transaction log and responds with "YES"; otherwise, it responds with "NO".

Second Phase (Commit/Rollback Phase)

All votes are YES: If all participants respond with "YES", the coordinator sends a "commit" command to all participants.
Some votes are NO: If any participant responds with "NO", or does not respond within a specified time, the coordinator sends a "rollback" command to all participants. Each participant, after receiving the "commit" or "rollback" command, performs the corresponding operation and sends a confirmation to the coordinator.

Pros and Cons

Pros:

Ensures transaction atomicity in a distributed environment: The two-phase commit protocol ensures transaction atomicity in a distributed environment through two stages of operation. In the first stage, all participants pre-execute the transaction and decide whether they can commit the transaction based on their local situation. In the second stage, if all participants agree to commit the transaction, the transaction will be committed, otherwise it will be rolled back. In this way, whether the transaction is committed or not, all participants' states can be consistent, thus achieving transaction atomicity.
The structure is relatively simple and easy to understand: The process and structure of the two-phase commit protocol are relatively simple. It only includes two stages: the preparation stage and the commit/rollback stage. In the preparation stage, the coordinator sends a transaction request to all participants, and the participants vote based on their local situation. In the commit/rollback stage, the coordinator decides whether to commit or rollback the transaction based on the voting results. This process is intuitive, easy to understand, and easy to implement.

Cons:

Performance overhead: The two-phase commit protocol involves multiple network communications, including the coordinator sending transaction requests to participants, participants returning voting results to the coordinator, and the coordinator sending commit or rollback commands to participants. These communication processes will increase the system latency, especially in poor network environments, the performance overhead may be greater.
Single point of failure: In the two-phase commit protocol, the coordinator plays a key role. If the coordinator fails or crashes, it will cause the participants to be in an uncertain state, not knowing whether to commit or rollback the transaction. In this case, the participants must wait for the coordinator to recover before they can continue to execute, which may cause the overall performance of the system to decline. Multiple coordinators can be used to avoid single point of failure. For example, you can use the master-slave mode. When the master coordinator fails, the backup coordinator can take over. In addition, heartbeat detection and timeout mechanisms can be used to detect and handle coordinator failures.
Blocking problem: In the second stage of the two-phase commit protocol, if a participant does not return a voting result, or the returned result is "NO", then the coordinator will send a rollback command to all participants. However, if a participant fails or is delayed at this stage, other participants must wait for this participant to recover before they can continue to execute, which may cause a blocking problem. In some scenarios, this blockage may last a long time, seriously affecting the performance and availability of the system. Introducing a timeout mechanism can solve the blocking problem. If the participant does not respond within a specified time, the coordinator can decide to rollback the transaction to avoid other participants waiting for a long time.

Three-Phase Commit

However, these strategies cannot completely solve the problems of the two-phase commit protocol. In order to better solve the distributed transaction problem, we propose the three-phase commit protocol (3PC). 3PC adds a timeout mechanism and a pre-commit stage to 2PC, which can better solve the single point of failure and blocking problems.

The three-phase commit introduces a new stage, making the entire protocol have the following three stages:

First Phase (CanCommit Phase)

Commit Inquiry: The coordinator sends a commit inquiry to all participants, asking them whether they can enter the next stage, that is, commit the transaction.
Participant Decision: Each participant decides whether it can commit based on its own status, and then responds with "can" or "cannot".

Second Phase (PreCommit Phase)

All participants agree: If all participants reply "can", then the coordinator decides to continue to commit and enter this stage, sending a "pre-commit" message to all participants.
Participant Pre-execution: After receiving the "pre-commit" message, the participant executes the transaction operation but does not commit, and then sends a "ready to commit" message to the coordinator.

Third Phase (DoCommit Phase)

All participants are ready: After the coordinator receives the "ready to commit" message from all participants, it sends them a "commit" message.
Participant Commit: After receiving the "commit" message, the participant officially commits its transaction.
Transaction Abort: If during this process, any participant or coordinator sends an "abort" message or does not respond, then the transaction will be aborted.

Pros:

Better solves single point of failure and blocking problems: The three-phase commit protocol adds a pre-commit stage and a timeout mechanism to the two-phase commit, which makes it better at solving single point of failure and blocking problems. In the pre-commit stage, participants execute transaction operations but do not commit, which provides more flexibility for the coordinator to handle possible failures and blockages.
Enhances system availability and robustness: By introducing a timeout mechanism, the three-phase commit protocol can avoid other participants waiting for a long time in the case of no response from a participant or a response of "NO", thereby enhancing the availability and robustness of the system.
Improves transaction execution efficiency: In the three-phase commit protocol, participants begin to execute transaction operations in the pre-commit stage, which can improve the execution efficiency of the transaction, especially when the transaction operation takes a long time.

Cons and Solutions:

Higher message overhead: The three-phase commit protocol adds a stage to the two-phase commit protocol, which means more message exchanges, leading to higher message overhead. This may affect the performance of the system, especially in poor network environments. The solution is to optimize the network environment and message transmission mechanism to reduce message overhead.
Increased complexity: The introduction of a new pre-commit stage and timeout mechanism increases the complexity of the protocol. This may increase the difficulty of implementation and maintenance. The solution is to use more advanced protocols, such as Paxos or Raft, which can better handle failures and blocking problems when processing distributed transactions, and also provide higher performance.
Blocking problem still exists: Although the three-phase commit protocol solves the blocking problem by introducing a timeout mechanism, in some cases, such as network partitioning or simultaneous failure of the coordinator and participants, blocking may still occur. The solution is to introduce a fault recovery mechanism, such as log recording. When the coordinator or participant fails, the state can be recovered through the log to avoid blocking.

Conclusion

In a distributed system, transaction processing is an important issue. Two-phase commit (2PC) and three-phase commit (3PC) protocols are two common solutions. 2PC ensures transaction atomicity in a distributed environment through two stages of operation, but it has performance overhead, single point of failure, and blocking problems. 3PC adds a pre-commit stage and a timeout mechanism to 2PC, better solves the single point of failure and blocking problems, improves the execution efficiency of the transaction, but brings higher message overhead and complexity. In practical applications, the appropriate protocol needs to be selected based on the specific requirements and environment of the system.