qqeasonchen opened a new pull request, #5206:
URL: https://github.com/apache/eventmesh/pull/5206
…ibe architecture
This comprehensive implementation introduces a complete A2A protocol for
EventMesh that enables intelligent multi-agent collaboration through a
publish/subscribe model instead of traditional point-to-point communication.
## Core Architecture
### 1. EventMesh-Native Publish/Subscribe Model
- A2APublishSubscribeService: Core service leveraging
EventMeshProducer/Consumer
- Anonymous task publishing without knowing specific consumer agents
- Topic-based routing (a2a.tasks.*, a2a.results, a2a.status)
- Integration with EventMesh storage plugins (RocketMQ, Kafka, Pulsar, Redis)
- CloudEvents 1.0 compliant message format
### 2. Protocol Infrastructure
- A2AProtocolAdaptor: Basic protocol adapter for A2A message processing
- EnhancedA2AProtocolAdaptor: Advanced adapter with protocol delegation
- EnhancedProtocolPluginFactory: High-performance factory with caching
- ProtocolRouter: Intelligent routing with rule-based message forwarding
- ProtocolMetrics: Comprehensive performance monitoring and statistics
### 3. Agent Management & Discovery
- AgentRegistry: Agent discovery and metadata management with heartbeat
monitoring
- Capability-based agent discovery and subscription matching
- Automatic agent lifecycle management and cleanup
- Agent health monitoring with configurable timeouts
### 4. Workflow Orchestration
- CollaborationManager: Multi-agent workflow orchestration using pub/sub
- Task-based workflow execution with dependency management
- Session management for complex multi-step processes
- Fault tolerance with automatic retry and recovery
### 5. Advanced Task Management
- Complete task lifecycle: Request → Message → Processing → Result
- Retry logic with exponential backoff and maximum attempt limits
- Task timeout handling and cancellation support
- Correlation ID tracking for workflow orchestration
- Priority-based task processing with multiple priority levels
## Key Features
### Publish/Subscribe Capabilities
- **Anonymous Publishing**: Publishers don't need to know consumers
- **Capability-Based Routing**: Tasks routed based on required capabilities
- **Automatic Load Balancing**: Multiple agents with same capabilities share
workload
- **Subscription Management**: Agents subscribe to task types they can handle
### EventMesh Integration
- **Storage Plugin Support**: Persistent message queues via EventMesh storage
- **Multi-Protocol Transport**: HTTP, gRPC, TCP protocol support
- **Event Streaming**: Real-time event streaming for monitoring
- **CloudEvents Standard**: Full CloudEvents 1.0 specification compliance
### Production Features
- **Fault Tolerance**: Automatic failover and retry mechanisms
- **Metrics & Monitoring**: Comprehensive performance tracking
- **Scalability**: Horizontal scaling through EventMesh topics
- **Observability**: Full visibility into task execution and agent status
## Implementation Components
### Protocol Layer
- EnhancedA2AProtocolAdaptor with protocol delegation
- CloudEvents conversion and message transformation
- Multi-protocol support (HTTP, gRPC, TCP)
### Runtime Services
- A2APublishSubscribeService for core pub/sub operations
- MessageRouter refactored for pub/sub delegation
- A2AMessageHandler for message processing
- A2AProtocolProcessor for protocol-level operations
### Management Services
- AgentRegistry for agent lifecycle management
- CollaborationManager for workflow orchestration
- SubscriptionRegistry for subscription management
- TaskMetricsCollector for performance monitoring
### Examples & Documentation
- Complete data processing pipeline demo
- Publish/subscribe usage examples
- Docker compose setup for testing
- Comprehensive documentation in English and Chinese
## Benefits Over Point-to-Point Model
- **True Horizontal Scalability**: EventMesh topics support unlimited scaling
- **Fault Tolerance**: Persistent queues with automatic retry and DLQ
- **Complete Decoupling**: Publishers and consumers operate independently
- **Load Distribution**: Automatic load balancing across agent pools
- **EventMesh Ecosystem**: Full integration with EventMesh infrastructure
- **Production Ready**: Enterprise-grade reliability and monitoring
## Usage Example
```java
// Publish task without knowing specific consumers
A2ATaskRequest taskRequest = A2ATaskRequest.builder()
.taskType("data-processing")
.payload(Map.of("data", "user-behavior"))
.requiredCapabilities(List.of("data-processing"))
.priority(A2ATaskPriority.HIGH)
.build();
pubSubService.publishTask(taskRequest);
// Subscribe to task types based on agent capabilities
pubSubService.subscribeToTaskType("agent-001", "data-processing",
List.of("data-processing", "analytics"), taskHandler);
```
This implementation transforms A2A from a simple agent communication
protocol into a production-ready, EventMesh-native multi-agent orchestration
platform suitable for large-scale distributed AI and automation systems.
<!--
### Contribution Checklist
- Name the pull request in the form "[ISSUE #XXXX] Title of the pull
request",
where *XXXX* should be replaced by the actual issue number.
Skip *[ISSUE #XXXX]* if there is no associated github issue for this
pull request.
- Fill out the template below to describe the changes contributed by the
pull request.
That will give reviewers the context they need to do the review.
- Each pull request should address only one issue.
Please do not mix up code from multiple issues.
- Each commit in the pull request should have a meaningful commit message.
- Once all items of the checklist are addressed, remove the above text and
this checklist,
leaving only the filled out template below.
(The sections below can be removed for hotfixes of typos)
-->
<!--
(If this PR fixes a GitHub issue, please add `Fixes #<XXX>` or `Closes
#<XXX>`.)
-->
Fixes #issue_id
### Motivation
*Explain the content here.*
*Explain why you want to make the changes and what problem you're trying to
solve.*
### Modifications
*Describe the modifications you've done.*
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs /
not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a followup
issue for adding the documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
