Copilot commented on code in PR #16008:
URL: https://github.com/apache/dubbo/pull/16008#discussion_r2697168126
##########
dubbo-demo/DUBBO_STREAM_BEST_PRACTICES.md:
##########
@@ -0,0 +1,1341 @@
+# Best Practices for Using Dubbo Stream
+
+## Table of Contents
+
+1. [Overview](#overview)
+2. [When to Use Dubbo Stream](#when-to-use-dubbo-stream)
+3. [Architecture Overview](#architecture-overview)
+4. [Stream Types and Patterns](#stream-types-and-patterns)
+5. [Handling Network Interruptions](#handling-network-interruptions)
+6. [Strong Consistency Strategies](#strong-consistency-strategies)
+7. [Idempotency Implementation](#idempotency-implementation)
+8. [Checkpoint Mechanisms](#checkpoint-mechanisms)
+9. [Error Handling and Recovery](#error-handling-and-recovery)
+10. [Configuration Best Practices](#configuration-best-practices)
+11. [Common Mistakes and Solutions](#common-mistakes-and-solutions)
+12. [Conclusion](#conclusion)
+
+---
+
+## Overview
+
+Dubbo Stream provides a modern, gRPC-compatible streaming communication model
for Apache Dubbo. It enables efficient bidirectional, server-to-client, and
client-to-server streaming patterns with full support for the Triple protocol,
making it ideal for handling high-volume data transfers, event-driven
architectures, and real-time communication scenarios.
+
+This guide outlines proven best practices for implementing streaming services
in Dubbo, with emphasis on reliability, consistency, and error resilience in
production environments.
+
+---
+
+## When to Use Dubbo Stream
+
+Dubbo Stream is particularly well-suited for the following scenarios:
+
+### 1. High-Volume Data Transfer
+When your application needs to transfer large datasets efficiently without
creating multiple individual RPC calls. Stream reduces serialization overhead
and connection costs.
+
+**Example:** Bulk data exports, file uploads/downloads, log aggregation.
+
+### 2. Real-Time Event Processing
+Applications that require continuous data flow with low latency between
services.
+
+**Example:** Real-time metrics collection, live data feeds, event
notifications.
+
+### 3. Bidirectional Communication
+Services where both client and server need to send data independently during a
single RPC call.
+
+**Example:** Real-time chat applications, collaborative tools, remote
procedure monitoring.
+
+### 4. Backpressure-Sensitive Operations
+Scenarios where the producer and consumer operate at different rates, and you
need to handle flow control gracefully.
+
+**Example:** Message processing pipelines, data aggregation services.
+
+---
+
+## Architecture Overview
+
+### Stream Communication Model
+
+Dubbo Stream supports three primary communication patterns through the Triple
protocol:
+
+```
+Client Server
+ | |
+ |-------- Server Stream (1-to-many) ------|
+ | |
+ |-------- Client Stream (many-to-1) ------|
+ | |
+ |---- Bidirectional Stream (many-many) ---|
+```
+
+### Core Components
+
+1. **StreamObserver**: The callback interface for handling stream events
+ - `onNext(T value)`: Receives the next item in the stream
+ - `onError(Throwable throwable)`: Handles errors
+ - `onCompleted()`: Signals stream completion
+
+2. **Service Definition**: Protobuf service interface defining stream contracts
+
+3. **Protocol Handler**: Triple protocol manages connection lifecycle,
serialization, and resource cleanup
+
+---
+
+## Stream Types and Patterns
+
+### 1. Server Stream (One Request, Many Responses)
+
+Server streams are useful when a client sends a single request and the server
responds with multiple items over time.
+
+**Service Definition (Protobuf):**
+```protobuf
+syntax = "proto3";
+
+package org.apache.dubbo.demo;
+
+message Request {
+ string query = 1;
+}
+
+message Response {
+ string data = 1;
+}
+
+service DataService {
+ rpc streamData(Request) returns (stream Response);
+}
+```
+
+**Server Implementation:**
+```java
+import org.apache.dubbo.common.stream.StreamObserver;
+
+public class DataServiceImpl implements DataService {
+
+ @Override
+ public void streamData(Request request,
+ StreamObserver<Response> responseObserver) {
+ try {
+ // Fetch data in batches
+ List<String> dataItems = fetchData(request.getQuery());
+
+ for (String item : dataItems) {
+ Response response = Response.newBuilder()
+ .setData(item)
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ private List<String> fetchData(String query) {
+ // Implementation
+ return new ArrayList<>();
+ }
+}
+```
+
+**Client Consumption:**
+```java
+DataService.Stub stub = DataService.newStub(channel);
+
+StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processData(value.getData());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error: ", t);
+ // Handle error and potentially retry
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Stream completed successfully");
+ }
+ };
+
+Request request = Request.newBuilder()
+ .setQuery("important_data")
+ .build();
+
+stub.streamData(request, responseObserver);
+```
+
+### 2. Client Stream (Many Requests, One Response)
+
+Client streams allow multiple items to be sent to the server before receiving
a single response.
+
+**Service Definition:**
+```protobuf
+service BatchService {
+ rpc processBatch(stream BatchItem) returns (Result);
+}
+
+message BatchItem {
+ int64 id = 1;
+ bytes data = 2;
+}
+
+message Result {
+ int32 processedCount = 1;
+ string status = 2;
+}
+```
+
+**Server Implementation:**
+```java
+public class BatchServiceImpl implements BatchService {
+
+ @Override
+ public StreamObserver<BatchItem> processBatch(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<BatchItem>() {
+ private int count = 0;
+
+ @Override
+ public void onNext(BatchItem value) {
+ try {
+ processSingleItem(value);
+ count++;
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Client stream error: ", t);
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ Result result = Result.newBuilder()
+ .setProcessedCount(count)
+ .setStatus("SUCCESS")
+ .build();
+ responseObserver.onNext(result);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processSingleItem(BatchItem item) {
+ // Implementation
+ }
+}
+```
+
+**Client Implementation:**
+```java
+BatchService.Stub stub = BatchService.newStub(channel);
+
+StreamObserver<Result> responseObserver =
+ new StreamObserver<Result>() {
+ @Override
+ public void onNext(Result value) {
+ logger.info("Processed {} items with status: {}",
+ value.getProcessedCount(),
+ value.getStatus());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Batch processing error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Batch completed");
+ }
+ };
+
+StreamObserver<BatchItem> requestObserver =
+ stub.processBatch(responseObserver);
+
+for (BatchItem item : itemsToProcess) {
+ requestObserver.onNext(item);
+}
+
+requestObserver.onCompleted();
+```
+
+### 3. Bidirectional Stream (Many Requests, Many Responses)
+
+Bidirectional streams enable independent request and response flows, useful
for interactive applications.
+
+**Service Definition:**
+```protobuf
+service ChatService {
+ rpc chat(stream ChatMessage) returns (stream ChatMessage);
+}
+
+message ChatMessage {
+ string sender = 1;
+ string content = 2;
+ int64 timestamp = 3;
+}
+```
+
+**Server Implementation:**
+```java
+public class ChatServiceImpl implements ChatService {
+
+ @Override
+ public StreamObserver<ChatMessage> chat(
+ StreamObserver<ChatMessage> responseObserver) {
+
+ return new StreamObserver<ChatMessage>() {
+ private String clientId = UUID.randomUUID().toString();
+
+ @Override
+ public void onNext(ChatMessage message) {
+ // Broadcast to all connected clients
+ broadcastMessage(message, clientId);
+
+ // Echo back with server processing
+ ChatMessage response = ChatMessage.newBuilder()
+ .setSender("SERVER")
+ .setContent("Received: " + message.getContent())
+ .setTimestamp(System.currentTimeMillis())
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Chat error for {}: ", clientId, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Chat session completed for {}", clientId);
+ disconnectClient(clientId);
+ }
+ };
+ }
+
+ private void broadcastMessage(ChatMessage message, String senderId) {
+ // Broadcast logic
+ }
+
+ private void disconnectClient(String clientId) {
+ // Cleanup logic
+ }
+}
+```
+
+---
+
+## Handling Network Interruptions
+
+Network instability is a reality in production. Proper handling of
interruptions is critical for stream reliability.
+
+### 1. Connection Timeout Configuration
+
+Configure appropriate timeouts in your Dubbo application:
+
+**YAML Configuration:**
+```yaml
+dubbo:
+ consumer:
+ timeout: 30000 # 30 seconds for unary calls
+ provider:
+ timeout: 30000
+ protocol:
+ name: triple
+ port: 50051
+ # Stream-specific settings
+ parameters:
+ keep-alive-time: 30000 # Milliseconds
+ keep-alive-timeout: 10000
+ max-idle-time: 600000 # 10 minutes
+```
+
+**Java Configuration:**
+```java
+@Configuration
+public class DubboConfig {
+
+ @Bean
+ public ServiceConfig<DataService> dataServiceConfig() {
+ ServiceConfig<DataService> config = new ServiceConfig<>();
+ config.setInterface(DataService.class);
+ config.setRef(new DataServiceImpl());
+ config.setProtocol(tripleProtocolConfig());
+ return config;
+ }
+
+ @Bean
+ public ProtocolConfig tripleProtocolConfig() {
+ ProtocolConfig config = new ProtocolConfig();
+ config.setName("triple");
+ config.setPort(50051);
+ config.getParameters().put("keep-alive-time", "30000");
+ config.getParameters().put("keep-alive-timeout", "10000");
+ return config;
+ }
+}
+```
+
+### 2. Graceful Degradation
+
+Implement fallback mechanisms for stream failures:
+
+```java
+public class ResilientDataService {
+
+ private final DataService.Stub primaryStub;
+ private final DataService.Stub fallbackStub;
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ public void streamDataWithFallback(Request request) {
+ final AtomicInteger errorCount = new AtomicInteger(0);
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processResponse(value);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ errorCount.incrementAndGet();
+ logger.warn("Primary stream failed, attempting fallback",
t);
+
+ if (errorCount.get() >= 3) {
+ logger.error("Fallback exhausted");
+ completionLatch.countDown();
+ return;
+ }
+
+ // Use fallback service
+ primaryStub.streamData(request, this);
Review Comment:
This fallback logic is incorrect. The code is calling
`primaryStub.streamData(request, this)` but should be using
`fallbackStub.streamData(request, this)` instead. The purpose of this error
handler is to attempt fallback to the secondary service when the primary fails,
but it's currently retrying with the same primary stub that already failed.
```suggestion
fallbackStub.streamData(request, this);
```
##########
dubbo-demo/DUBBO_STREAM_BEST_PRACTICES.md:
##########
@@ -0,0 +1,1341 @@
+# Best Practices for Using Dubbo Stream
+
+## Table of Contents
+
+1. [Overview](#overview)
+2. [When to Use Dubbo Stream](#when-to-use-dubbo-stream)
+3. [Architecture Overview](#architecture-overview)
+4. [Stream Types and Patterns](#stream-types-and-patterns)
+5. [Handling Network Interruptions](#handling-network-interruptions)
+6. [Strong Consistency Strategies](#strong-consistency-strategies)
+7. [Idempotency Implementation](#idempotency-implementation)
+8. [Checkpoint Mechanisms](#checkpoint-mechanisms)
+9. [Error Handling and Recovery](#error-handling-and-recovery)
+10. [Configuration Best Practices](#configuration-best-practices)
+11. [Common Mistakes and Solutions](#common-mistakes-and-solutions)
+12. [Conclusion](#conclusion)
+
+---
+
+## Overview
+
+Dubbo Stream provides a modern, gRPC-compatible streaming communication model
for Apache Dubbo. It enables efficient bidirectional, server-to-client, and
client-to-server streaming patterns with full support for the Triple protocol,
making it ideal for handling high-volume data transfers, event-driven
architectures, and real-time communication scenarios.
+
+This guide outlines proven best practices for implementing streaming services
in Dubbo, with emphasis on reliability, consistency, and error resilience in
production environments.
+
+---
+
+## When to Use Dubbo Stream
+
+Dubbo Stream is particularly well-suited for the following scenarios:
+
+### 1. High-Volume Data Transfer
+When your application needs to transfer large datasets efficiently without
creating multiple individual RPC calls. Stream reduces serialization overhead
and connection costs.
+
+**Example:** Bulk data exports, file uploads/downloads, log aggregation.
+
+### 2. Real-Time Event Processing
+Applications that require continuous data flow with low latency between
services.
+
+**Example:** Real-time metrics collection, live data feeds, event
notifications.
+
+### 3. Bidirectional Communication
+Services where both client and server need to send data independently during a
single RPC call.
+
+**Example:** Real-time chat applications, collaborative tools, remote
procedure monitoring.
+
+### 4. Backpressure-Sensitive Operations
+Scenarios where the producer and consumer operate at different rates, and you
need to handle flow control gracefully.
+
+**Example:** Message processing pipelines, data aggregation services.
+
+---
+
+## Architecture Overview
+
+### Stream Communication Model
+
+Dubbo Stream supports three primary communication patterns through the Triple
protocol:
+
+```
+Client Server
+ | |
+ |-------- Server Stream (1-to-many) ------|
+ | |
+ |-------- Client Stream (many-to-1) ------|
+ | |
+ |---- Bidirectional Stream (many-many) ---|
+```
+
+### Core Components
+
+1. **StreamObserver**: The callback interface for handling stream events
+ - `onNext(T value)`: Receives the next item in the stream
+ - `onError(Throwable throwable)`: Handles errors
+ - `onCompleted()`: Signals stream completion
+
+2. **Service Definition**: Protobuf service interface defining stream contracts
+
+3. **Protocol Handler**: Triple protocol manages connection lifecycle,
serialization, and resource cleanup
+
+---
+
+## Stream Types and Patterns
+
+### 1. Server Stream (One Request, Many Responses)
+
+Server streams are useful when a client sends a single request and the server
responds with multiple items over time.
+
+**Service Definition (Protobuf):**
+```protobuf
+syntax = "proto3";
+
+package org.apache.dubbo.demo;
+
+message Request {
+ string query = 1;
+}
+
+message Response {
+ string data = 1;
+}
+
+service DataService {
+ rpc streamData(Request) returns (stream Response);
+}
+```
+
+**Server Implementation:**
+```java
+import org.apache.dubbo.common.stream.StreamObserver;
+
+public class DataServiceImpl implements DataService {
+
+ @Override
+ public void streamData(Request request,
+ StreamObserver<Response> responseObserver) {
+ try {
+ // Fetch data in batches
+ List<String> dataItems = fetchData(request.getQuery());
+
+ for (String item : dataItems) {
+ Response response = Response.newBuilder()
+ .setData(item)
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ private List<String> fetchData(String query) {
+ // Implementation
+ return new ArrayList<>();
+ }
+}
+```
+
+**Client Consumption:**
+```java
+DataService.Stub stub = DataService.newStub(channel);
+
+StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processData(value.getData());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error: ", t);
+ // Handle error and potentially retry
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Stream completed successfully");
+ }
+ };
+
+Request request = Request.newBuilder()
+ .setQuery("important_data")
+ .build();
+
+stub.streamData(request, responseObserver);
+```
+
+### 2. Client Stream (Many Requests, One Response)
+
+Client streams allow multiple items to be sent to the server before receiving
a single response.
+
+**Service Definition:**
+```protobuf
+service BatchService {
+ rpc processBatch(stream BatchItem) returns (Result);
+}
+
+message BatchItem {
+ int64 id = 1;
+ bytes data = 2;
+}
+
+message Result {
+ int32 processedCount = 1;
+ string status = 2;
+}
+```
+
+**Server Implementation:**
+```java
+public class BatchServiceImpl implements BatchService {
+
+ @Override
+ public StreamObserver<BatchItem> processBatch(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<BatchItem>() {
+ private int count = 0;
+
+ @Override
+ public void onNext(BatchItem value) {
+ try {
+ processSingleItem(value);
+ count++;
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Client stream error: ", t);
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ Result result = Result.newBuilder()
+ .setProcessedCount(count)
+ .setStatus("SUCCESS")
+ .build();
+ responseObserver.onNext(result);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processSingleItem(BatchItem item) {
+ // Implementation
+ }
+}
+```
+
+**Client Implementation:**
+```java
+BatchService.Stub stub = BatchService.newStub(channel);
+
+StreamObserver<Result> responseObserver =
+ new StreamObserver<Result>() {
+ @Override
+ public void onNext(Result value) {
+ logger.info("Processed {} items with status: {}",
+ value.getProcessedCount(),
+ value.getStatus());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Batch processing error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Batch completed");
+ }
+ };
+
+StreamObserver<BatchItem> requestObserver =
+ stub.processBatch(responseObserver);
+
+for (BatchItem item : itemsToProcess) {
+ requestObserver.onNext(item);
+}
+
+requestObserver.onCompleted();
+```
+
+### 3. Bidirectional Stream (Many Requests, Many Responses)
+
+Bidirectional streams enable independent request and response flows, useful
for interactive applications.
+
+**Service Definition:**
+```protobuf
+service ChatService {
+ rpc chat(stream ChatMessage) returns (stream ChatMessage);
+}
+
+message ChatMessage {
+ string sender = 1;
+ string content = 2;
+ int64 timestamp = 3;
+}
+```
+
+**Server Implementation:**
+```java
+public class ChatServiceImpl implements ChatService {
+
+ @Override
+ public StreamObserver<ChatMessage> chat(
+ StreamObserver<ChatMessage> responseObserver) {
+
+ return new StreamObserver<ChatMessage>() {
+ private String clientId = UUID.randomUUID().toString();
+
+ @Override
+ public void onNext(ChatMessage message) {
+ // Broadcast to all connected clients
+ broadcastMessage(message, clientId);
+
+ // Echo back with server processing
+ ChatMessage response = ChatMessage.newBuilder()
+ .setSender("SERVER")
+ .setContent("Received: " + message.getContent())
+ .setTimestamp(System.currentTimeMillis())
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Chat error for {}: ", clientId, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Chat session completed for {}", clientId);
+ disconnectClient(clientId);
+ }
+ };
+ }
+
+ private void broadcastMessage(ChatMessage message, String senderId) {
+ // Broadcast logic
+ }
+
+ private void disconnectClient(String clientId) {
+ // Cleanup logic
+ }
+}
+```
+
+---
+
+## Handling Network Interruptions
+
+Network instability is a reality in production. Proper handling of
interruptions is critical for stream reliability.
+
+### 1. Connection Timeout Configuration
+
+Configure appropriate timeouts in your Dubbo application:
+
+**YAML Configuration:**
+```yaml
+dubbo:
+ consumer:
+ timeout: 30000 # 30 seconds for unary calls
+ provider:
+ timeout: 30000
+ protocol:
+ name: triple
+ port: 50051
+ # Stream-specific settings
+ parameters:
+ keep-alive-time: 30000 # Milliseconds
+ keep-alive-timeout: 10000
+ max-idle-time: 600000 # 10 minutes
+```
+
+**Java Configuration:**
+```java
+@Configuration
+public class DubboConfig {
+
+ @Bean
+ public ServiceConfig<DataService> dataServiceConfig() {
+ ServiceConfig<DataService> config = new ServiceConfig<>();
+ config.setInterface(DataService.class);
+ config.setRef(new DataServiceImpl());
+ config.setProtocol(tripleProtocolConfig());
+ return config;
+ }
+
+ @Bean
+ public ProtocolConfig tripleProtocolConfig() {
+ ProtocolConfig config = new ProtocolConfig();
+ config.setName("triple");
+ config.setPort(50051);
+ config.getParameters().put("keep-alive-time", "30000");
+ config.getParameters().put("keep-alive-timeout", "10000");
+ return config;
+ }
+}
+```
+
+### 2. Graceful Degradation
+
+Implement fallback mechanisms for stream failures:
+
+```java
+public class ResilientDataService {
+
+ private final DataService.Stub primaryStub;
+ private final DataService.Stub fallbackStub;
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ public void streamDataWithFallback(Request request) {
+ final AtomicInteger errorCount = new AtomicInteger(0);
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processResponse(value);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ errorCount.incrementAndGet();
+ logger.warn("Primary stream failed, attempting fallback",
t);
+
+ if (errorCount.get() >= 3) {
+ logger.error("Fallback exhausted");
+ completionLatch.countDown();
+ return;
+ }
+
+ // Use fallback service
+ primaryStub.streamData(request, this);
+ }
+
+ @Override
+ public void onCompleted() {
+ completionLatch.countDown();
+ }
+ };
+
+ primaryStub.streamData(request, responseObserver);
+
+ try {
+ completionLatch.await(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void processResponse(Response value) {
+ // Process response
+ }
+}
+```
+
+### 3. Heartbeat and Keepalive Mechanism
+
+Maintain stream connectivity with periodic heartbeats:
+
+```java
+public class HeartbeatStreamService {
+
+ @Override
+ public StreamObserver<StreamMessage> streamWithHeartbeat(
+ StreamObserver<StreamMessage> responseObserver) {
+
+ ScheduledExecutorService executor =
+ Executors.newScheduledThreadPool(1);
+
+ return new StreamObserver<StreamMessage>() {
+ private volatile long lastActivityTime =
+ System.currentTimeMillis();
+
+ @Override
+ public void onNext(StreamMessage message) {
+ lastActivityTime = System.currentTimeMillis();
+ processMessage(message);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ executor.shutdownNow();
+ }
+
+ @Override
+ public void onCompleted() {
+ executor.shutdownNow();
+ responseObserver.onCompleted();
+ }
+ };
Review Comment:
The heartbeat implementation is incomplete. A ScheduledExecutorService is
created but no periodic heartbeat task is actually scheduled. The
`lastActivityTime` field is updated but never checked to detect inactivity. A
complete implementation should schedule a periodic task that checks the last
activity time and sends heartbeat messages or detects timeouts. Consider adding
something like `executor.scheduleAtFixedRate(() -> checkHeartbeat(), 0, 30,
TimeUnit.SECONDS)` and implementing the heartbeat logic.
##########
dubbo-demo/DUBBO_STREAM_BEST_PRACTICES.md:
##########
@@ -0,0 +1,1341 @@
+# Best Practices for Using Dubbo Stream
+
+## Table of Contents
+
+1. [Overview](#overview)
+2. [When to Use Dubbo Stream](#when-to-use-dubbo-stream)
+3. [Architecture Overview](#architecture-overview)
+4. [Stream Types and Patterns](#stream-types-and-patterns)
+5. [Handling Network Interruptions](#handling-network-interruptions)
+6. [Strong Consistency Strategies](#strong-consistency-strategies)
+7. [Idempotency Implementation](#idempotency-implementation)
+8. [Checkpoint Mechanisms](#checkpoint-mechanisms)
+9. [Error Handling and Recovery](#error-handling-and-recovery)
+10. [Configuration Best Practices](#configuration-best-practices)
+11. [Common Mistakes and Solutions](#common-mistakes-and-solutions)
+12. [Conclusion](#conclusion)
+
+---
+
+## Overview
+
+Dubbo Stream provides a modern, gRPC-compatible streaming communication model
for Apache Dubbo. It enables efficient bidirectional, server-to-client, and
client-to-server streaming patterns with full support for the Triple protocol,
making it ideal for handling high-volume data transfers, event-driven
architectures, and real-time communication scenarios.
+
+This guide outlines proven best practices for implementing streaming services
in Dubbo, with emphasis on reliability, consistency, and error resilience in
production environments.
+
+---
+
+## When to Use Dubbo Stream
+
+Dubbo Stream is particularly well-suited for the following scenarios:
+
+### 1. High-Volume Data Transfer
+When your application needs to transfer large datasets efficiently without
creating multiple individual RPC calls. Stream reduces serialization overhead
and connection costs.
+
+**Example:** Bulk data exports, file uploads/downloads, log aggregation.
+
+### 2. Real-Time Event Processing
+Applications that require continuous data flow with low latency between
services.
+
+**Example:** Real-time metrics collection, live data feeds, event
notifications.
+
+### 3. Bidirectional Communication
+Services where both client and server need to send data independently during a
single RPC call.
+
+**Example:** Real-time chat applications, collaborative tools, remote
procedure monitoring.
+
+### 4. Backpressure-Sensitive Operations
+Scenarios where the producer and consumer operate at different rates, and you
need to handle flow control gracefully.
+
+**Example:** Message processing pipelines, data aggregation services.
+
+---
+
+## Architecture Overview
+
+### Stream Communication Model
+
+Dubbo Stream supports three primary communication patterns through the Triple
protocol:
+
+```
+Client Server
+ | |
+ |-------- Server Stream (1-to-many) ------|
+ | |
+ |-------- Client Stream (many-to-1) ------|
+ | |
+ |---- Bidirectional Stream (many-many) ---|
+```
+
+### Core Components
+
+1. **StreamObserver**: The callback interface for handling stream events
+ - `onNext(T value)`: Receives the next item in the stream
+ - `onError(Throwable throwable)`: Handles errors
+ - `onCompleted()`: Signals stream completion
+
+2. **Service Definition**: Protobuf service interface defining stream contracts
+
+3. **Protocol Handler**: Triple protocol manages connection lifecycle,
serialization, and resource cleanup
+
+---
+
+## Stream Types and Patterns
+
+### 1. Server Stream (One Request, Many Responses)
+
+Server streams are useful when a client sends a single request and the server
responds with multiple items over time.
+
+**Service Definition (Protobuf):**
+```protobuf
+syntax = "proto3";
+
+package org.apache.dubbo.demo;
+
+message Request {
+ string query = 1;
+}
+
+message Response {
+ string data = 1;
+}
+
+service DataService {
+ rpc streamData(Request) returns (stream Response);
+}
+```
+
+**Server Implementation:**
+```java
+import org.apache.dubbo.common.stream.StreamObserver;
+
+public class DataServiceImpl implements DataService {
+
+ @Override
+ public void streamData(Request request,
+ StreamObserver<Response> responseObserver) {
+ try {
+ // Fetch data in batches
+ List<String> dataItems = fetchData(request.getQuery());
+
+ for (String item : dataItems) {
+ Response response = Response.newBuilder()
+ .setData(item)
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ private List<String> fetchData(String query) {
+ // Implementation
+ return new ArrayList<>();
+ }
+}
+```
+
+**Client Consumption:**
+```java
+DataService.Stub stub = DataService.newStub(channel);
+
+StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processData(value.getData());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error: ", t);
+ // Handle error and potentially retry
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Stream completed successfully");
+ }
+ };
+
+Request request = Request.newBuilder()
+ .setQuery("important_data")
+ .build();
+
+stub.streamData(request, responseObserver);
+```
+
+### 2. Client Stream (Many Requests, One Response)
+
+Client streams allow multiple items to be sent to the server before receiving
a single response.
+
+**Service Definition:**
+```protobuf
+service BatchService {
+ rpc processBatch(stream BatchItem) returns (Result);
+}
+
+message BatchItem {
+ int64 id = 1;
+ bytes data = 2;
+}
+
+message Result {
+ int32 processedCount = 1;
+ string status = 2;
+}
+```
+
+**Server Implementation:**
+```java
+public class BatchServiceImpl implements BatchService {
+
+ @Override
+ public StreamObserver<BatchItem> processBatch(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<BatchItem>() {
+ private int count = 0;
+
+ @Override
+ public void onNext(BatchItem value) {
+ try {
+ processSingleItem(value);
+ count++;
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Client stream error: ", t);
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ Result result = Result.newBuilder()
+ .setProcessedCount(count)
+ .setStatus("SUCCESS")
+ .build();
+ responseObserver.onNext(result);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processSingleItem(BatchItem item) {
+ // Implementation
+ }
+}
+```
+
+**Client Implementation:**
+```java
+BatchService.Stub stub = BatchService.newStub(channel);
+
+StreamObserver<Result> responseObserver =
+ new StreamObserver<Result>() {
+ @Override
+ public void onNext(Result value) {
+ logger.info("Processed {} items with status: {}",
+ value.getProcessedCount(),
+ value.getStatus());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Batch processing error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Batch completed");
+ }
+ };
+
+StreamObserver<BatchItem> requestObserver =
+ stub.processBatch(responseObserver);
+
+for (BatchItem item : itemsToProcess) {
+ requestObserver.onNext(item);
+}
+
+requestObserver.onCompleted();
+```
+
+### 3. Bidirectional Stream (Many Requests, Many Responses)
+
+Bidirectional streams enable independent request and response flows, useful
for interactive applications.
+
+**Service Definition:**
+```protobuf
+service ChatService {
+ rpc chat(stream ChatMessage) returns (stream ChatMessage);
+}
+
+message ChatMessage {
+ string sender = 1;
+ string content = 2;
+ int64 timestamp = 3;
+}
+```
+
+**Server Implementation:**
+```java
+public class ChatServiceImpl implements ChatService {
+
+ @Override
+ public StreamObserver<ChatMessage> chat(
+ StreamObserver<ChatMessage> responseObserver) {
+
+ return new StreamObserver<ChatMessage>() {
+ private String clientId = UUID.randomUUID().toString();
+
+ @Override
+ public void onNext(ChatMessage message) {
+ // Broadcast to all connected clients
+ broadcastMessage(message, clientId);
+
+ // Echo back with server processing
+ ChatMessage response = ChatMessage.newBuilder()
+ .setSender("SERVER")
+ .setContent("Received: " + message.getContent())
+ .setTimestamp(System.currentTimeMillis())
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Chat error for {}: ", clientId, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Chat session completed for {}", clientId);
+ disconnectClient(clientId);
+ }
+ };
+ }
+
+ private void broadcastMessage(ChatMessage message, String senderId) {
+ // Broadcast logic
+ }
+
+ private void disconnectClient(String clientId) {
+ // Cleanup logic
+ }
+}
+```
+
+---
+
+## Handling Network Interruptions
+
+Network instability is a reality in production. Proper handling of
interruptions is critical for stream reliability.
+
+### 1. Connection Timeout Configuration
+
+Configure appropriate timeouts in your Dubbo application:
+
+**YAML Configuration:**
+```yaml
+dubbo:
+ consumer:
+ timeout: 30000 # 30 seconds for unary calls
+ provider:
+ timeout: 30000
+ protocol:
+ name: triple
+ port: 50051
+ # Stream-specific settings
+ parameters:
+ keep-alive-time: 30000 # Milliseconds
+ keep-alive-timeout: 10000
+ max-idle-time: 600000 # 10 minutes
+```
+
+**Java Configuration:**
+```java
+@Configuration
+public class DubboConfig {
+
+ @Bean
+ public ServiceConfig<DataService> dataServiceConfig() {
+ ServiceConfig<DataService> config = new ServiceConfig<>();
+ config.setInterface(DataService.class);
+ config.setRef(new DataServiceImpl());
+ config.setProtocol(tripleProtocolConfig());
+ return config;
+ }
+
+ @Bean
+ public ProtocolConfig tripleProtocolConfig() {
+ ProtocolConfig config = new ProtocolConfig();
+ config.setName("triple");
+ config.setPort(50051);
+ config.getParameters().put("keep-alive-time", "30000");
+ config.getParameters().put("keep-alive-timeout", "10000");
+ return config;
+ }
+}
+```
+
+### 2. Graceful Degradation
+
+Implement fallback mechanisms for stream failures:
+
+```java
+public class ResilientDataService {
+
+ private final DataService.Stub primaryStub;
+ private final DataService.Stub fallbackStub;
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ public void streamDataWithFallback(Request request) {
+ final AtomicInteger errorCount = new AtomicInteger(0);
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processResponse(value);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ errorCount.incrementAndGet();
+ logger.warn("Primary stream failed, attempting fallback",
t);
+
+ if (errorCount.get() >= 3) {
+ logger.error("Fallback exhausted");
+ completionLatch.countDown();
+ return;
+ }
+
+ // Use fallback service
+ primaryStub.streamData(request, this);
+ }
+
+ @Override
+ public void onCompleted() {
+ completionLatch.countDown();
+ }
+ };
+
+ primaryStub.streamData(request, responseObserver);
+
+ try {
+ completionLatch.await(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void processResponse(Response value) {
+ // Process response
+ }
+}
+```
+
+### 3. Heartbeat and Keepalive Mechanism
+
+Maintain stream connectivity with periodic heartbeats:
+
+```java
+public class HeartbeatStreamService {
+
+ @Override
+ public StreamObserver<StreamMessage> streamWithHeartbeat(
+ StreamObserver<StreamMessage> responseObserver) {
+
+ ScheduledExecutorService executor =
+ Executors.newScheduledThreadPool(1);
+
+ return new StreamObserver<StreamMessage>() {
+ private volatile long lastActivityTime =
+ System.currentTimeMillis();
+
+ @Override
+ public void onNext(StreamMessage message) {
+ lastActivityTime = System.currentTimeMillis();
+ processMessage(message);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ executor.shutdownNow();
+ }
+
+ @Override
+ public void onCompleted() {
+ executor.shutdownNow();
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processMessage(StreamMessage message) {
+ // Implementation
+ }
+}
+```
+
+---
+
+## Strong Consistency Strategies
+
+Achieving strong consistency in streaming scenarios requires careful
coordination between client and server.
+
+### 1. Message Ordering Guarantees
+
+Ensure messages are processed in order:
+
+```java
+public class OrderedStreamProcessor {
+
+ private final AtomicLong lastProcessedSequence =
+ new AtomicLong(-1);
+
+ public void processOrderedStream(
+ StreamMessage message,
+ StreamObserver<StreamMessage> responseObserver) {
+
+ long currentSequence = message.getSequence();
+ long expectedSequence = lastProcessedSequence.get() + 1;
+
+ if (currentSequence == expectedSequence) {
+ // Process message
+ processMessage(message);
+ lastProcessedSequence.set(currentSequence);
+
+ // Send acknowledgment
+ StreamMessage ack = StreamMessage.newBuilder()
+ .setSequence(currentSequence)
+ .setType("ACK")
+ .build();
+ responseObserver.onNext(ack);
+
+ } else if (currentSequence > expectedSequence) {
+ // Out of order - request retransmission
+ StreamMessage nack = StreamMessage.newBuilder()
+ .setSequence(expectedSequence)
+ .setType("NACK")
+ .build();
+ responseObserver.onNext(nack);
+
+ } else {
+ // Duplicate - send ACK but don't process
+ StreamMessage ack = StreamMessage.newBuilder()
+ .setSequence(currentSequence)
+ .setType("ACK")
+ .build();
+ responseObserver.onNext(ack);
+ }
+ }
+
+ private void processMessage(StreamMessage message) {
+ // Implementation
+ }
+}
+```
+
+### 2. Transactional Stream Processing
+
+Batch messages and process them transactionally:
+
+```java
+public class TransactionalStreamService {
+
+ private final int batchSize = 100;
+ private final List<StreamMessage> batch = new ArrayList<>();
+ private final Object lock = new Object();
+
+ @Override
+ public StreamObserver<StreamMessage> processTransactional(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<StreamMessage>() {
+
+ @Override
+ public void onNext(StreamMessage message) {
+ synchronized (lock) {
+ batch.add(message);
+
+ if (batch.size() >= batchSize) {
+ processBatch();
+ }
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ synchronized (lock) {
+ // Rollback uncommitted batch
+ batch.clear();
+ }
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ synchronized (lock) {
+ if (!batch.isEmpty()) {
+ processBatch();
+ }
+ }
+ responseObserver.onCompleted();
+ }
+
+ private void processBatch() {
+ try {
+ // Execute in transaction
+ executeTransaction(batch);
+
+ Result result = Result.newBuilder()
+ .setSuccess(true)
+ .setCount(batch.size())
+ .build();
+ responseObserver.onNext(result);
+
+ batch.clear();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+ };
+ }
+
+ private void executeTransaction(List<StreamMessage> messages) {
+ // Transaction implementation
+ }
+}
+```
+
+---
+
+## Idempotency Implementation
+
+Idempotency ensures that duplicate messages (often due to retries) don't cause
unintended side effects.
+
+### 1. Message ID Tracking
+
+Use message IDs to detect and handle duplicates:
+
+```java
+public class IdempotentStreamProcessor {
+
+ private final Set<String> processedMessageIds =
+ Collections.synchronizedSet(new HashSet<>());
Review Comment:
This in-memory Set will grow unbounded as more unique message IDs are
processed, potentially causing memory issues in long-running applications.
Consider using a bounded cache with eviction policies (e.g., Caffeine cache
with time-based or size-based eviction) or the persistent approach shown in the
next section. This is similar to the unbounded buffer issue mentioned in the
Common Mistakes section.
##########
dubbo-demo/DUBBO_STREAM_BEST_PRACTICES.md:
##########
@@ -0,0 +1,1341 @@
+# Best Practices for Using Dubbo Stream
+
+## Table of Contents
+
+1. [Overview](#overview)
+2. [When to Use Dubbo Stream](#when-to-use-dubbo-stream)
+3. [Architecture Overview](#architecture-overview)
+4. [Stream Types and Patterns](#stream-types-and-patterns)
+5. [Handling Network Interruptions](#handling-network-interruptions)
+6. [Strong Consistency Strategies](#strong-consistency-strategies)
+7. [Idempotency Implementation](#idempotency-implementation)
+8. [Checkpoint Mechanisms](#checkpoint-mechanisms)
+9. [Error Handling and Recovery](#error-handling-and-recovery)
+10. [Configuration Best Practices](#configuration-best-practices)
+11. [Common Mistakes and Solutions](#common-mistakes-and-solutions)
+12. [Conclusion](#conclusion)
+
+---
+
+## Overview
+
+Dubbo Stream provides a modern, gRPC-compatible streaming communication model
for Apache Dubbo. It enables efficient bidirectional, server-to-client, and
client-to-server streaming patterns with full support for the Triple protocol,
making it ideal for handling high-volume data transfers, event-driven
architectures, and real-time communication scenarios.
+
+This guide outlines proven best practices for implementing streaming services
in Dubbo, with emphasis on reliability, consistency, and error resilience in
production environments.
+
+---
+
+## When to Use Dubbo Stream
+
+Dubbo Stream is particularly well-suited for the following scenarios:
+
+### 1. High-Volume Data Transfer
+When your application needs to transfer large datasets efficiently without
creating multiple individual RPC calls. Stream reduces serialization overhead
and connection costs.
+
+**Example:** Bulk data exports, file uploads/downloads, log aggregation.
+
+### 2. Real-Time Event Processing
+Applications that require continuous data flow with low latency between
services.
+
+**Example:** Real-time metrics collection, live data feeds, event
notifications.
+
+### 3. Bidirectional Communication
+Services where both client and server need to send data independently during a
single RPC call.
+
+**Example:** Real-time chat applications, collaborative tools, remote
procedure monitoring.
+
+### 4. Backpressure-Sensitive Operations
+Scenarios where the producer and consumer operate at different rates, and you
need to handle flow control gracefully.
+
+**Example:** Message processing pipelines, data aggregation services.
+
+---
+
+## Architecture Overview
+
+### Stream Communication Model
+
+Dubbo Stream supports three primary communication patterns through the Triple
protocol:
+
+```
+Client Server
+ | |
+ |-------- Server Stream (1-to-many) ------|
+ | |
+ |-------- Client Stream (many-to-1) ------|
+ | |
+ |---- Bidirectional Stream (many-many) ---|
+```
+
+### Core Components
+
+1. **StreamObserver**: The callback interface for handling stream events
+ - `onNext(T value)`: Receives the next item in the stream
+ - `onError(Throwable throwable)`: Handles errors
+ - `onCompleted()`: Signals stream completion
+
+2. **Service Definition**: Protobuf service interface defining stream contracts
+
+3. **Protocol Handler**: Triple protocol manages connection lifecycle,
serialization, and resource cleanup
+
+---
+
+## Stream Types and Patterns
+
+### 1. Server Stream (One Request, Many Responses)
+
+Server streams are useful when a client sends a single request and the server
responds with multiple items over time.
+
+**Service Definition (Protobuf):**
+```protobuf
+syntax = "proto3";
+
+package org.apache.dubbo.demo;
+
+message Request {
+ string query = 1;
+}
+
+message Response {
+ string data = 1;
+}
+
+service DataService {
+ rpc streamData(Request) returns (stream Response);
+}
+```
+
+**Server Implementation:**
+```java
+import org.apache.dubbo.common.stream.StreamObserver;
+
+public class DataServiceImpl implements DataService {
+
+ @Override
+ public void streamData(Request request,
+ StreamObserver<Response> responseObserver) {
+ try {
+ // Fetch data in batches
+ List<String> dataItems = fetchData(request.getQuery());
+
+ for (String item : dataItems) {
+ Response response = Response.newBuilder()
+ .setData(item)
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ private List<String> fetchData(String query) {
+ // Implementation
+ return new ArrayList<>();
+ }
+}
+```
+
+**Client Consumption:**
+```java
+DataService.Stub stub = DataService.newStub(channel);
+
+StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processData(value.getData());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error: ", t);
+ // Handle error and potentially retry
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Stream completed successfully");
+ }
+ };
+
+Request request = Request.newBuilder()
+ .setQuery("important_data")
+ .build();
+
+stub.streamData(request, responseObserver);
+```
+
+### 2. Client Stream (Many Requests, One Response)
+
+Client streams allow multiple items to be sent to the server before receiving
a single response.
+
+**Service Definition:**
+```protobuf
+service BatchService {
+ rpc processBatch(stream BatchItem) returns (Result);
+}
+
+message BatchItem {
+ int64 id = 1;
+ bytes data = 2;
+}
+
+message Result {
+ int32 processedCount = 1;
+ string status = 2;
+}
+```
+
+**Server Implementation:**
+```java
+public class BatchServiceImpl implements BatchService {
+
+ @Override
+ public StreamObserver<BatchItem> processBatch(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<BatchItem>() {
+ private int count = 0;
+
+ @Override
+ public void onNext(BatchItem value) {
+ try {
+ processSingleItem(value);
+ count++;
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Client stream error: ", t);
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ Result result = Result.newBuilder()
+ .setProcessedCount(count)
+ .setStatus("SUCCESS")
+ .build();
+ responseObserver.onNext(result);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processSingleItem(BatchItem item) {
+ // Implementation
+ }
+}
+```
+
+**Client Implementation:**
+```java
+BatchService.Stub stub = BatchService.newStub(channel);
+
+StreamObserver<Result> responseObserver =
+ new StreamObserver<Result>() {
+ @Override
+ public void onNext(Result value) {
+ logger.info("Processed {} items with status: {}",
+ value.getProcessedCount(),
+ value.getStatus());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Batch processing error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Batch completed");
+ }
+ };
+
+StreamObserver<BatchItem> requestObserver =
+ stub.processBatch(responseObserver);
+
+for (BatchItem item : itemsToProcess) {
+ requestObserver.onNext(item);
+}
+
+requestObserver.onCompleted();
+```
+
+### 3. Bidirectional Stream (Many Requests, Many Responses)
+
+Bidirectional streams enable independent request and response flows, useful
for interactive applications.
+
+**Service Definition:**
+```protobuf
+service ChatService {
+ rpc chat(stream ChatMessage) returns (stream ChatMessage);
+}
+
+message ChatMessage {
+ string sender = 1;
+ string content = 2;
+ int64 timestamp = 3;
+}
+```
+
+**Server Implementation:**
+```java
+public class ChatServiceImpl implements ChatService {
+
+ @Override
+ public StreamObserver<ChatMessage> chat(
+ StreamObserver<ChatMessage> responseObserver) {
+
+ return new StreamObserver<ChatMessage>() {
+ private String clientId = UUID.randomUUID().toString();
+
+ @Override
+ public void onNext(ChatMessage message) {
+ // Broadcast to all connected clients
+ broadcastMessage(message, clientId);
+
+ // Echo back with server processing
+ ChatMessage response = ChatMessage.newBuilder()
+ .setSender("SERVER")
+ .setContent("Received: " + message.getContent())
+ .setTimestamp(System.currentTimeMillis())
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Chat error for {}: ", clientId, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Chat session completed for {}", clientId);
+ disconnectClient(clientId);
+ }
+ };
+ }
+
+ private void broadcastMessage(ChatMessage message, String senderId) {
+ // Broadcast logic
+ }
+
+ private void disconnectClient(String clientId) {
+ // Cleanup logic
+ }
+}
+```
+
+---
+
+## Handling Network Interruptions
+
+Network instability is a reality in production. Proper handling of
interruptions is critical for stream reliability.
+
+### 1. Connection Timeout Configuration
+
+Configure appropriate timeouts in your Dubbo application:
+
+**YAML Configuration:**
+```yaml
+dubbo:
+ consumer:
+ timeout: 30000 # 30 seconds for unary calls
+ provider:
+ timeout: 30000
+ protocol:
+ name: triple
+ port: 50051
+ # Stream-specific settings
+ parameters:
+ keep-alive-time: 30000 # Milliseconds
+ keep-alive-timeout: 10000
+ max-idle-time: 600000 # 10 minutes
+```
+
+**Java Configuration:**
+```java
+@Configuration
+public class DubboConfig {
+
+ @Bean
+ public ServiceConfig<DataService> dataServiceConfig() {
+ ServiceConfig<DataService> config = new ServiceConfig<>();
+ config.setInterface(DataService.class);
+ config.setRef(new DataServiceImpl());
+ config.setProtocol(tripleProtocolConfig());
+ return config;
+ }
+
+ @Bean
+ public ProtocolConfig tripleProtocolConfig() {
+ ProtocolConfig config = new ProtocolConfig();
+ config.setName("triple");
+ config.setPort(50051);
+ config.getParameters().put("keep-alive-time", "30000");
+ config.getParameters().put("keep-alive-timeout", "10000");
+ return config;
+ }
+}
+```
+
+### 2. Graceful Degradation
+
+Implement fallback mechanisms for stream failures:
+
+```java
+public class ResilientDataService {
+
+ private final DataService.Stub primaryStub;
+ private final DataService.Stub fallbackStub;
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ public void streamDataWithFallback(Request request) {
+ final AtomicInteger errorCount = new AtomicInteger(0);
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processResponse(value);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ errorCount.incrementAndGet();
+ logger.warn("Primary stream failed, attempting fallback",
t);
+
+ if (errorCount.get() >= 3) {
+ logger.error("Fallback exhausted");
+ completionLatch.countDown();
+ return;
+ }
+
+ // Use fallback service
+ primaryStub.streamData(request, this);
+ }
+
+ @Override
+ public void onCompleted() {
+ completionLatch.countDown();
+ }
+ };
+
+ primaryStub.streamData(request, responseObserver);
+
+ try {
+ completionLatch.await(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void processResponse(Response value) {
+ // Process response
+ }
+}
+```
+
+### 3. Heartbeat and Keepalive Mechanism
+
+Maintain stream connectivity with periodic heartbeats:
+
+```java
+public class HeartbeatStreamService {
+
+ @Override
+ public StreamObserver<StreamMessage> streamWithHeartbeat(
+ StreamObserver<StreamMessage> responseObserver) {
+
+ ScheduledExecutorService executor =
+ Executors.newScheduledThreadPool(1);
+
+ return new StreamObserver<StreamMessage>() {
+ private volatile long lastActivityTime =
+ System.currentTimeMillis();
+
+ @Override
+ public void onNext(StreamMessage message) {
+ lastActivityTime = System.currentTimeMillis();
+ processMessage(message);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ executor.shutdownNow();
+ }
+
+ @Override
+ public void onCompleted() {
+ executor.shutdownNow();
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processMessage(StreamMessage message) {
+ // Implementation
+ }
+}
+```
+
+---
+
+## Strong Consistency Strategies
+
+Achieving strong consistency in streaming scenarios requires careful
coordination between client and server.
+
+### 1. Message Ordering Guarantees
+
+Ensure messages are processed in order:
+
+```java
+public class OrderedStreamProcessor {
+
+ private final AtomicLong lastProcessedSequence =
+ new AtomicLong(-1);
+
+ public void processOrderedStream(
+ StreamMessage message,
+ StreamObserver<StreamMessage> responseObserver) {
+
+ long currentSequence = message.getSequence();
+ long expectedSequence = lastProcessedSequence.get() + 1;
+
+ if (currentSequence == expectedSequence) {
+ // Process message
+ processMessage(message);
+ lastProcessedSequence.set(currentSequence);
+
+ // Send acknowledgment
+ StreamMessage ack = StreamMessage.newBuilder()
+ .setSequence(currentSequence)
+ .setType("ACK")
+ .build();
+ responseObserver.onNext(ack);
+
+ } else if (currentSequence > expectedSequence) {
+ // Out of order - request retransmission
+ StreamMessage nack = StreamMessage.newBuilder()
+ .setSequence(expectedSequence)
+ .setType("NACK")
+ .build();
+ responseObserver.onNext(nack);
+
+ } else {
+ // Duplicate - send ACK but don't process
+ StreamMessage ack = StreamMessage.newBuilder()
+ .setSequence(currentSequence)
+ .setType("ACK")
+ .build();
+ responseObserver.onNext(ack);
+ }
+ }
+
+ private void processMessage(StreamMessage message) {
+ // Implementation
+ }
+}
+```
+
+### 2. Transactional Stream Processing
+
+Batch messages and process them transactionally:
+
+```java
+public class TransactionalStreamService {
+
+ private final int batchSize = 100;
+ private final List<StreamMessage> batch = new ArrayList<>();
+ private final Object lock = new Object();
+
+ @Override
+ public StreamObserver<StreamMessage> processTransactional(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<StreamMessage>() {
+
+ @Override
+ public void onNext(StreamMessage message) {
+ synchronized (lock) {
+ batch.add(message);
+
+ if (batch.size() >= batchSize) {
+ processBatch();
+ }
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ synchronized (lock) {
+ // Rollback uncommitted batch
+ batch.clear();
+ }
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ synchronized (lock) {
+ if (!batch.isEmpty()) {
+ processBatch();
+ }
+ }
+ responseObserver.onCompleted();
+ }
+
+ private void processBatch() {
+ try {
+ // Execute in transaction
+ executeTransaction(batch);
+
+ Result result = Result.newBuilder()
+ .setSuccess(true)
+ .setCount(batch.size())
+ .build();
+ responseObserver.onNext(result);
+
+ batch.clear();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+ };
+ }
+
+ private void executeTransaction(List<StreamMessage> messages) {
+ // Transaction implementation
+ }
+}
+```
+
+---
+
+## Idempotency Implementation
+
+Idempotency ensures that duplicate messages (often due to retries) don't cause
unintended side effects.
+
+### 1. Message ID Tracking
+
+Use message IDs to detect and handle duplicates:
+
+```java
+public class IdempotentStreamProcessor {
+
+ private final Set<String> processedMessageIds =
+ Collections.synchronizedSet(new HashSet<>());
+
+ @Override
+ public StreamObserver<Message> idempotentStream(
+ StreamObserver<Response> responseObserver) {
+
+ return new StreamObserver<Message>() {
+
+ @Override
+ public void onNext(Message message) {
+ String messageId = message.getId();
+
+ if (processedMessageIds.contains(messageId)) {
+ // Duplicate - send previous response
+ Response response = getCachedResponse(messageId);
+ responseObserver.onNext(response);
+ return;
+ }
+
+ // Process message
+ try {
+ Response response = processAndCacheResult(message);
+ processedMessageIds.add(messageId);
+ responseObserver.onNext(response);
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private Response processAndCacheResult(Message message) {
+ // Implementation
+ return Response.newBuilder().build();
+ }
+
+ private Response getCachedResponse(String messageId) {
+ // Retrieve from cache
+ return Response.newBuilder().build();
+ }
+}
+```
+
+### 2. Idempotency Key Storage
+
+Persist idempotency keys for longer-term deduplication:
+
+```java
+public class PersistentIdempotencyService {
+
+ private final IdempotencyKeyRepository repository;
+
+ @Override
+ public StreamObserver<Operation> persistentIdempotentOps(
+ StreamObserver<OperationResult> responseObserver) {
+
+ return new StreamObserver<Operation>() {
+
+ @Override
+ public void onNext(Operation op) {
+ String idempotencyKey = op.getIdempotencyKey();
+
+ // Check database for existing result
+ Optional<OperationResult> cachedResult =
+ repository.findByIdempotencyKey(idempotencyKey);
+
+ if (cachedResult.isPresent()) {
+ responseObserver.onNext(cachedResult.get());
+ return;
+ }
+
+ try {
+ OperationResult result = executeOperation(op);
+
+ // Store result with idempotency key
+ repository.save(new IdempotencyRecord(
+ idempotencyKey,
+ result,
+ System.currentTimeMillis()
+ ));
+
+ responseObserver.onNext(result);
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Operation stream error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private OperationResult executeOperation(Operation op) {
+ // Implementation
+ return OperationResult.newBuilder().build();
+ }
+}
+```
+
+---
+
+## Checkpoint Mechanisms
+
+Checkpoints enable resuming streams from the last known good state after
interruptions.
+
+### 1. Periodic Checkpoint Recording
+
+Record progress at regular intervals:
+
+```java
+public class CheckpointedStreamProcessor {
+
+ private final CheckpointRepository repository;
+ private volatile long lastCheckpoint = 0;
+ private final long checkpointInterval = 1000; // 1 second
+
+ @Override
+ public StreamObserver<DataChunk> streamWithCheckpoints(
+ String streamId,
+ StreamObserver<CheckpointAck> responseObserver) {
+
+ return new StreamObserver<DataChunk>() {
+ private long processedBytes = 0;
+ private long lastCheckpointTime = System.currentTimeMillis();
+
+ @Override
+ public void onNext(DataChunk chunk) {
+ try {
+ processChunk(chunk);
+ processedBytes += chunk.getData().size();
+
+ // Record checkpoint periodically
+ long now = System.currentTimeMillis();
+ if (now - lastCheckpointTime >= checkpointInterval) {
+ recordCheckpoint(streamId, processedBytes);
+
+ CheckpointAck ack = CheckpointAck.newBuilder()
+ .setStreamId(streamId)
+ .setCheckpointOffset(processedBytes)
+ .setTimestamp(now)
+ .build();
+ responseObserver.onNext(ack);
+
+ lastCheckpointTime = now;
+ }
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error at byte offset {}: ",
+ processedBytes, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ recordCheckpoint(streamId, processedBytes);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void recordCheckpoint(String streamId, long offset) {
+ repository.saveCheckpoint(
+ new Checkpoint(streamId, offset, System.currentTimeMillis())
+ );
+ }
+
+ private void processChunk(DataChunk chunk) {
+ // Implementation
+ }
+}
+```
+
+### 2. Stream Resume from Checkpoint
+
+Resume streaming from last checkpoint:
+
+```java
+public class CheckpointResumeService {
+
+ private final CheckpointRepository repository;
+
+ public void resumeStream(String streamId,
+ StreamObserver<DataChunk> responseObserver) {
+
+ // Retrieve last checkpoint
+ Optional<Checkpoint> lastCheckpoint =
+ repository.getLatestCheckpoint(streamId);
+
+ long resumeFromOffset = lastCheckpoint
+ .map(Checkpoint::getOffset)
+ .orElse(0L);
+
+ logger.info("Resuming stream {} from offset {}",
+ streamId, resumeFromOffset);
+
+ try {
+ // Stream data starting from checkpoint
+ streamDataFromOffset(streamId, resumeFromOffset,
+ responseObserver);
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ private void streamDataFromOffset(String streamId,
+ long offset,
+ StreamObserver<DataChunk> observer) {
+ // Implementation: fetch and stream data from offset
+ }
+}
+```
+
+---
+
+## Error Handling and Recovery
+
+Robust error handling is essential for production stream services.
+
+### 1. Comprehensive Error Categorization
+
+Differentiate between recoverable and fatal errors:
+
+```java
+public class ErrorHandlingStreamService {
+
+ public enum ErrorType {
+ TRANSIENT, // Retry recommended
+ PERMANENT, // Don't retry
+ DEGRADED // Continue with reduced functionality
+ }
+
+ @Override
+ public StreamObserver<Request> robustStream(
+ StreamObserver<Response> responseObserver) {
+
+ return new StreamObserver<Request>() {
+ private int retryCount = 0;
+ private static final int MAX_RETRIES = 3;
+
+ @Override
+ public void onNext(Request request) {
+ try {
+ processRequest(request, responseObserver);
+ retryCount = 0; // Reset on success
+ } catch (Exception e) {
+ handleStreamError(e, responseObserver);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ ErrorType errorType = classifyError(t);
+
+ switch (errorType) {
+ case TRANSIENT:
+ if (retryCount < MAX_RETRIES) {
+ retryCount++;
+ logger.info("Transient error, retry {} of {}",
+ retryCount, MAX_RETRIES);
+ // Retry logic
+ } else {
+ logger.error("Max retries exceeded");
+ responseObserver.onError(t);
+ }
+ break;
+
+ case PERMANENT:
+ logger.error("Permanent error, aborting stream", t);
+ responseObserver.onError(t);
+ break;
+
+ case DEGRADED:
+ logger.warn("Degraded mode", t);
+ // Continue with degraded service
+ break;
+ }
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+
+ private void processRequest(Request request,
+ StreamObserver<Response> observer) {
+ // Implementation
+ }
+
+ private void handleStreamError(Exception e,
+ StreamObserver<Response> observer) {
+ // Implementation
+ }
+
+ private ErrorType classifyError(Throwable t) {
+ if (t instanceof TimeoutException ||
+ t instanceof SocketTimeoutException) {
+ return ErrorType.TRANSIENT;
+ } else if (t instanceof IllegalArgumentException) {
+ return ErrorType.PERMANENT;
+ }
+ return ErrorType.DEGRADED;
+ }
+ };
+ }
+}
+```
+
+### 2. Circuit Breaker Pattern
+
+Implement circuit breaker for failing services:
+
+```java
+public class CircuitBreakerStreamService {
+
+ public enum CircuitState {
+ CLOSED, // Normal operation
+ OPEN, // Fail fast
+ HALF_OPEN // Testing recovery
+ }
+
+ private volatile CircuitState state = CircuitState.CLOSED;
+ private volatile long lastFailureTime = 0;
+ private volatile int failureCount = 0;
+ private static final int FAILURE_THRESHOLD = 5;
+ private static final long RESET_TIMEOUT = 30000; // 30 seconds
+
+ public StreamObserver<Request> streamWithCircuitBreaker(
+ StreamObserver<Response> responseObserver) {
+
+ return new StreamObserver<Request>() {
+
+ @Override
+ public void onNext(Request request) {
+ if (state == CircuitState.OPEN) {
+ if (System.currentTimeMillis() - lastFailureTime
+ > RESET_TIMEOUT) {
+ state = CircuitState.HALF_OPEN;
+ logger.info("Circuit breaker: HALF_OPEN");
+ } else {
+ responseObserver.onError(
+ new RuntimeException("Circuit breaker OPEN"));
+ return;
+ }
+ }
+
+ try {
+ processRequest(request, responseObserver);
+
+ if (state == CircuitState.HALF_OPEN) {
+ state = CircuitState.CLOSED;
+ failureCount = 0;
+ logger.info("Circuit breaker: CLOSED");
+ }
+ } catch (Exception e) {
+ failureCount++;
+ lastFailureTime = System.currentTimeMillis();
+
+ if (failureCount >= FAILURE_THRESHOLD) {
+ state = CircuitState.OPEN;
+ logger.warn("Circuit breaker: OPEN");
+ }
+
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Circuit breaker error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+
+ private void processRequest(Request request,
+ StreamObserver<Response> observer) {
+ // Implementation
+ }
+ };
+ }
+}
+```
+
+---
+
+## Configuration Best Practices
+
+### 1. Production Configuration Example
+
+**application.yml:**
+```yaml
+dubbo:
+ application:
+ name: dubbo-stream-service
+ version: 1.0.0
+
+ protocol:
+ name: triple
+ port: 50051
+ # Stream-specific settings
+ parameters:
+ keep-alive-time: 30000
+ keep-alive-timeout: 10000
+ max-idle-time: 600000
+ # Thread pool configuration
+ threads: 200
+ queues: 500
+ core-threads: 100
+
+ provider:
+ timeout: 30000
+ retries: 0 # No retries for stream
+ validation: true
+ serialization: protobuf
+
+ consumer:
+ timeout: 30000
+ # Client-side settings
+ pool-core-size: 10
+ pool-max-size: 100
+
+ registry:
+ address: zookeeper://localhost:2181
+
+ config-center:
+ protocol: zookeeper
+ address: localhost:2181
+
+ metrics:
+ protocol: prometheus
+ port: 9090
+
+ tracing:
+ name: zipkin
+ address: localhost:9411
+```
+
+### 2. Resource Management Configuration
+
+```java
+@Configuration
+public class StreamResourceConfig {
+
+ @Bean
+ public ExecutorService streamExecutor() {
+ return new ThreadPoolExecutor(
+ 100, // core threads
+ 200, // max threads
+ 60, // keep-alive time
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(500),
+ new ThreadFactory() {
+ private final AtomicInteger count = new AtomicInteger(0);
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r);
+ t.setName("stream-handler-" + count.incrementAndGet());
+ t.setDaemon(false);
+ return t;
+ }
+ }
+ );
+ }
+
+ @Bean
+ public StreamMetricsRegistry metricsRegistry() {
+ return new StreamMetricsRegistry();
+ }
+}
+```
+
+### 3. Monitoring and Observability Configuration
+
+```java
+@Configuration
+public class MonitoringConfig {
+
+ @Bean
+ public MeterRegistry meterRegistry() {
+ PrometheusMeterRegistry registry =
+ new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+
+ // Stream-specific meters
+ registry.counter("stream.messages.sent", "service");
+ registry.counter("stream.messages.received", "service");
+ registry.counter("stream.errors", "service", "type");
+ registry.timer("stream.processing.duration", "service");
+ registry.gauge("stream.active.connections", "service");
+
+ return registry;
+ }
+
Review Comment:
The metric registration calls (counter, timer, gauge) create metric
instances but don't capture or store them for later use. In
Micrometer/Prometheus, you typically need to capture the returned metric
instances to increment counters or record values. Consider storing these as
fields or using them directly when recording metrics. For example: `Counter
sentCounter = registry.counter("stream.messages.sent", "service");` and then
later calling `sentCounter.increment()`.
```suggestion
// Store metric instances for reuse when recording metrics
private Counter sentCounter;
private Counter receivedCounter;
private Counter errorCounter;
private Timer processingTimer;
private Gauge activeConnectionsGauge;
@Bean
public MeterRegistry meterRegistry() {
PrometheusMeterRegistry registry =
new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
// Stream-specific meters
sentCounter = registry.counter("stream.messages.sent", "service");
receivedCounter = registry.counter("stream.messages.received",
"service");
errorCounter = registry.counter("stream.errors", "service", "type");
processingTimer = registry.timer("stream.processing.duration",
"service");
// Gauge bound to a value source method (placeholder implementation)
activeConnectionsGauge = Gauge
.builder("stream.active.connections", this,
MonitoringConfig::getActiveConnections)
.tag("service", "stream-service")
.register(registry);
return registry;
}
// Placeholder for actual active connections metric source
private double getActiveConnections() {
// Implement logic to return the current number of active connections
return 0.0;
}
```
##########
dubbo-demo/DUBBO_STREAM_BEST_PRACTICES.md:
##########
@@ -0,0 +1,1341 @@
+# Best Practices for Using Dubbo Stream
+
+## Table of Contents
+
+1. [Overview](#overview)
+2. [When to Use Dubbo Stream](#when-to-use-dubbo-stream)
+3. [Architecture Overview](#architecture-overview)
+4. [Stream Types and Patterns](#stream-types-and-patterns)
+5. [Handling Network Interruptions](#handling-network-interruptions)
+6. [Strong Consistency Strategies](#strong-consistency-strategies)
+7. [Idempotency Implementation](#idempotency-implementation)
+8. [Checkpoint Mechanisms](#checkpoint-mechanisms)
+9. [Error Handling and Recovery](#error-handling-and-recovery)
+10. [Configuration Best Practices](#configuration-best-practices)
+11. [Common Mistakes and Solutions](#common-mistakes-and-solutions)
+12. [Conclusion](#conclusion)
+
+---
+
+## Overview
+
+Dubbo Stream provides a modern, gRPC-compatible streaming communication model
for Apache Dubbo. It enables efficient bidirectional, server-to-client, and
client-to-server streaming patterns with full support for the Triple protocol,
making it ideal for handling high-volume data transfers, event-driven
architectures, and real-time communication scenarios.
+
+This guide outlines proven best practices for implementing streaming services
in Dubbo, with emphasis on reliability, consistency, and error resilience in
production environments.
+
+---
+
+## When to Use Dubbo Stream
+
+Dubbo Stream is particularly well-suited for the following scenarios:
+
+### 1. High-Volume Data Transfer
+When your application needs to transfer large datasets efficiently without
creating multiple individual RPC calls. Stream reduces serialization overhead
and connection costs.
+
+**Example:** Bulk data exports, file uploads/downloads, log aggregation.
+
+### 2. Real-Time Event Processing
+Applications that require continuous data flow with low latency between
services.
+
+**Example:** Real-time metrics collection, live data feeds, event
notifications.
+
+### 3. Bidirectional Communication
+Services where both client and server need to send data independently during a
single RPC call.
+
+**Example:** Real-time chat applications, collaborative tools, remote
procedure monitoring.
+
+### 4. Backpressure-Sensitive Operations
+Scenarios where the producer and consumer operate at different rates, and you
need to handle flow control gracefully.
+
+**Example:** Message processing pipelines, data aggregation services.
+
+---
+
+## Architecture Overview
+
+### Stream Communication Model
+
+Dubbo Stream supports three primary communication patterns through the Triple
protocol:
+
+```
+Client Server
+ | |
+ |-------- Server Stream (1-to-many) ------|
+ | |
+ |-------- Client Stream (many-to-1) ------|
+ | |
+ |---- Bidirectional Stream (many-many) ---|
+```
+
+### Core Components
+
+1. **StreamObserver**: The callback interface for handling stream events
+ - `onNext(T value)`: Receives the next item in the stream
+ - `onError(Throwable throwable)`: Handles errors
+ - `onCompleted()`: Signals stream completion
+
+2. **Service Definition**: Protobuf service interface defining stream contracts
+
+3. **Protocol Handler**: Triple protocol manages connection lifecycle,
serialization, and resource cleanup
+
+---
+
+## Stream Types and Patterns
+
+### 1. Server Stream (One Request, Many Responses)
+
+Server streams are useful when a client sends a single request and the server
responds with multiple items over time.
+
+**Service Definition (Protobuf):**
+```protobuf
+syntax = "proto3";
+
+package org.apache.dubbo.demo;
+
+message Request {
+ string query = 1;
+}
+
+message Response {
+ string data = 1;
+}
+
+service DataService {
+ rpc streamData(Request) returns (stream Response);
+}
+```
+
+**Server Implementation:**
+```java
+import org.apache.dubbo.common.stream.StreamObserver;
+
+public class DataServiceImpl implements DataService {
+
+ @Override
+ public void streamData(Request request,
+ StreamObserver<Response> responseObserver) {
+ try {
+ // Fetch data in batches
+ List<String> dataItems = fetchData(request.getQuery());
+
+ for (String item : dataItems) {
+ Response response = Response.newBuilder()
+ .setData(item)
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ private List<String> fetchData(String query) {
+ // Implementation
+ return new ArrayList<>();
+ }
+}
+```
+
+**Client Consumption:**
+```java
+DataService.Stub stub = DataService.newStub(channel);
+
+StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processData(value.getData());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error: ", t);
+ // Handle error and potentially retry
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Stream completed successfully");
+ }
+ };
+
+Request request = Request.newBuilder()
+ .setQuery("important_data")
+ .build();
+
+stub.streamData(request, responseObserver);
+```
+
+### 2. Client Stream (Many Requests, One Response)
+
+Client streams allow multiple items to be sent to the server before receiving
a single response.
+
+**Service Definition:**
+```protobuf
+service BatchService {
+ rpc processBatch(stream BatchItem) returns (Result);
+}
+
+message BatchItem {
+ int64 id = 1;
+ bytes data = 2;
+}
+
+message Result {
+ int32 processedCount = 1;
+ string status = 2;
+}
+```
+
+**Server Implementation:**
+```java
+public class BatchServiceImpl implements BatchService {
+
+ @Override
+ public StreamObserver<BatchItem> processBatch(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<BatchItem>() {
+ private int count = 0;
+
+ @Override
+ public void onNext(BatchItem value) {
+ try {
+ processSingleItem(value);
+ count++;
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Client stream error: ", t);
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ Result result = Result.newBuilder()
+ .setProcessedCount(count)
+ .setStatus("SUCCESS")
+ .build();
+ responseObserver.onNext(result);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processSingleItem(BatchItem item) {
+ // Implementation
+ }
+}
+```
+
+**Client Implementation:**
+```java
+BatchService.Stub stub = BatchService.newStub(channel);
+
+StreamObserver<Result> responseObserver =
+ new StreamObserver<Result>() {
+ @Override
+ public void onNext(Result value) {
+ logger.info("Processed {} items with status: {}",
+ value.getProcessedCount(),
+ value.getStatus());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Batch processing error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Batch completed");
+ }
+ };
+
+StreamObserver<BatchItem> requestObserver =
+ stub.processBatch(responseObserver);
+
+for (BatchItem item : itemsToProcess) {
+ requestObserver.onNext(item);
+}
+
+requestObserver.onCompleted();
+```
+
+### 3. Bidirectional Stream (Many Requests, Many Responses)
+
+Bidirectional streams enable independent request and response flows, useful
for interactive applications.
+
+**Service Definition:**
+```protobuf
+service ChatService {
+ rpc chat(stream ChatMessage) returns (stream ChatMessage);
+}
+
+message ChatMessage {
+ string sender = 1;
+ string content = 2;
+ int64 timestamp = 3;
+}
+```
+
+**Server Implementation:**
+```java
+public class ChatServiceImpl implements ChatService {
+
+ @Override
+ public StreamObserver<ChatMessage> chat(
+ StreamObserver<ChatMessage> responseObserver) {
+
+ return new StreamObserver<ChatMessage>() {
+ private String clientId = UUID.randomUUID().toString();
+
+ @Override
+ public void onNext(ChatMessage message) {
+ // Broadcast to all connected clients
+ broadcastMessage(message, clientId);
+
+ // Echo back with server processing
+ ChatMessage response = ChatMessage.newBuilder()
+ .setSender("SERVER")
+ .setContent("Received: " + message.getContent())
+ .setTimestamp(System.currentTimeMillis())
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Chat error for {}: ", clientId, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Chat session completed for {}", clientId);
+ disconnectClient(clientId);
+ }
+ };
+ }
+
+ private void broadcastMessage(ChatMessage message, String senderId) {
+ // Broadcast logic
+ }
+
+ private void disconnectClient(String clientId) {
+ // Cleanup logic
+ }
+}
+```
+
+---
+
+## Handling Network Interruptions
+
+Network instability is a reality in production. Proper handling of
interruptions is critical for stream reliability.
+
+### 1. Connection Timeout Configuration
+
+Configure appropriate timeouts in your Dubbo application:
+
+**YAML Configuration:**
+```yaml
+dubbo:
+ consumer:
+ timeout: 30000 # 30 seconds for unary calls
+ provider:
+ timeout: 30000
+ protocol:
+ name: triple
+ port: 50051
+ # Stream-specific settings
+ parameters:
+ keep-alive-time: 30000 # Milliseconds
+ keep-alive-timeout: 10000
+ max-idle-time: 600000 # 10 minutes
+```
+
+**Java Configuration:**
+```java
+@Configuration
+public class DubboConfig {
+
+ @Bean
+ public ServiceConfig<DataService> dataServiceConfig() {
+ ServiceConfig<DataService> config = new ServiceConfig<>();
+ config.setInterface(DataService.class);
+ config.setRef(new DataServiceImpl());
+ config.setProtocol(tripleProtocolConfig());
+ return config;
+ }
+
+ @Bean
+ public ProtocolConfig tripleProtocolConfig() {
+ ProtocolConfig config = new ProtocolConfig();
+ config.setName("triple");
+ config.setPort(50051);
+ config.getParameters().put("keep-alive-time", "30000");
+ config.getParameters().put("keep-alive-timeout", "10000");
+ return config;
+ }
+}
+```
+
+### 2. Graceful Degradation
+
+Implement fallback mechanisms for stream failures:
+
+```java
+public class ResilientDataService {
+
+ private final DataService.Stub primaryStub;
+ private final DataService.Stub fallbackStub;
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ public void streamDataWithFallback(Request request) {
+ final AtomicInteger errorCount = new AtomicInteger(0);
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processResponse(value);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ errorCount.incrementAndGet();
+ logger.warn("Primary stream failed, attempting fallback",
t);
+
+ if (errorCount.get() >= 3) {
+ logger.error("Fallback exhausted");
+ completionLatch.countDown();
+ return;
+ }
+
+ // Use fallback service
+ primaryStub.streamData(request, this);
+ }
+
+ @Override
+ public void onCompleted() {
+ completionLatch.countDown();
+ }
+ };
+
+ primaryStub.streamData(request, responseObserver);
+
+ try {
+ completionLatch.await(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void processResponse(Response value) {
+ // Process response
+ }
+}
+```
+
+### 3. Heartbeat and Keepalive Mechanism
+
+Maintain stream connectivity with periodic heartbeats:
+
+```java
+public class HeartbeatStreamService {
+
+ @Override
+ public StreamObserver<StreamMessage> streamWithHeartbeat(
+ StreamObserver<StreamMessage> responseObserver) {
+
+ ScheduledExecutorService executor =
+ Executors.newScheduledThreadPool(1);
+
+ return new StreamObserver<StreamMessage>() {
+ private volatile long lastActivityTime =
+ System.currentTimeMillis();
+
+ @Override
+ public void onNext(StreamMessage message) {
+ lastActivityTime = System.currentTimeMillis();
+ processMessage(message);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ executor.shutdownNow();
+ }
+
+ @Override
+ public void onCompleted() {
+ executor.shutdownNow();
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processMessage(StreamMessage message) {
+ // Implementation
+ }
+}
+```
+
+---
+
+## Strong Consistency Strategies
+
+Achieving strong consistency in streaming scenarios requires careful
coordination between client and server.
+
+### 1. Message Ordering Guarantees
+
+Ensure messages are processed in order:
+
+```java
+public class OrderedStreamProcessor {
+
+ private final AtomicLong lastProcessedSequence =
+ new AtomicLong(-1);
+
+ public void processOrderedStream(
+ StreamMessage message,
+ StreamObserver<StreamMessage> responseObserver) {
+
+ long currentSequence = message.getSequence();
+ long expectedSequence = lastProcessedSequence.get() + 1;
+
+ if (currentSequence == expectedSequence) {
+ // Process message
+ processMessage(message);
+ lastProcessedSequence.set(currentSequence);
+
+ // Send acknowledgment
+ StreamMessage ack = StreamMessage.newBuilder()
+ .setSequence(currentSequence)
+ .setType("ACK")
+ .build();
+ responseObserver.onNext(ack);
+
Review Comment:
This message ordering implementation has a race condition. Between checking
if currentSequence matches expectedSequence (line 515) and setting
lastProcessedSequence (line 518), another thread could process a different
message and update lastProcessedSequence, leading to messages being processed
out of order or skipped. Use AtomicLong.compareAndSet() to atomically check and
update the sequence number. For example: `if
(lastProcessedSequence.compareAndSet(expectedSequence - 1, currentSequence)) {
processMessage(message); ... }`
```suggestion
// Atomically advance the sequence to ensure ordered processing
if (lastProcessedSequence.compareAndSet(expectedSequence - 1,
currentSequence)) {
// Process message
processMessage(message);
// Send acknowledgment
StreamMessage ack = StreamMessage.newBuilder()
.setSequence(currentSequence)
.setType("ACK")
.build();
responseObserver.onNext(ack);
} else {
// Sequence already advanced by another thread; treat as
duplicate
StreamMessage ack = StreamMessage.newBuilder()
.setSequence(currentSequence)
.setType("ACK")
.build();
responseObserver.onNext(ack);
}
```
##########
dubbo-demo/DUBBO_STREAM_BEST_PRACTICES.md:
##########
@@ -0,0 +1,1341 @@
+# Best Practices for Using Dubbo Stream
+
+## Table of Contents
+
+1. [Overview](#overview)
+2. [When to Use Dubbo Stream](#when-to-use-dubbo-stream)
+3. [Architecture Overview](#architecture-overview)
+4. [Stream Types and Patterns](#stream-types-and-patterns)
+5. [Handling Network Interruptions](#handling-network-interruptions)
+6. [Strong Consistency Strategies](#strong-consistency-strategies)
+7. [Idempotency Implementation](#idempotency-implementation)
+8. [Checkpoint Mechanisms](#checkpoint-mechanisms)
+9. [Error Handling and Recovery](#error-handling-and-recovery)
+10. [Configuration Best Practices](#configuration-best-practices)
+11. [Common Mistakes and Solutions](#common-mistakes-and-solutions)
+12. [Conclusion](#conclusion)
+
+---
+
+## Overview
+
+Dubbo Stream provides a modern, gRPC-compatible streaming communication model
for Apache Dubbo. It enables efficient bidirectional, server-to-client, and
client-to-server streaming patterns with full support for the Triple protocol,
making it ideal for handling high-volume data transfers, event-driven
architectures, and real-time communication scenarios.
+
+This guide outlines proven best practices for implementing streaming services
in Dubbo, with emphasis on reliability, consistency, and error resilience in
production environments.
+
+---
+
+## When to Use Dubbo Stream
+
+Dubbo Stream is particularly well-suited for the following scenarios:
+
+### 1. High-Volume Data Transfer
+When your application needs to transfer large datasets efficiently without
creating multiple individual RPC calls. Stream reduces serialization overhead
and connection costs.
+
+**Example:** Bulk data exports, file uploads/downloads, log aggregation.
+
+### 2. Real-Time Event Processing
+Applications that require continuous data flow with low latency between
services.
+
+**Example:** Real-time metrics collection, live data feeds, event
notifications.
+
+### 3. Bidirectional Communication
+Services where both client and server need to send data independently during a
single RPC call.
+
+**Example:** Real-time chat applications, collaborative tools, remote
procedure monitoring.
+
+### 4. Backpressure-Sensitive Operations
+Scenarios where the producer and consumer operate at different rates, and you
need to handle flow control gracefully.
+
+**Example:** Message processing pipelines, data aggregation services.
+
+---
+
+## Architecture Overview
+
+### Stream Communication Model
+
+Dubbo Stream supports three primary communication patterns through the Triple
protocol:
+
+```
+Client Server
+ | |
+ |-------- Server Stream (1-to-many) ------|
+ | |
+ |-------- Client Stream (many-to-1) ------|
+ | |
+ |---- Bidirectional Stream (many-many) ---|
+```
+
+### Core Components
+
+1. **StreamObserver**: The callback interface for handling stream events
+ - `onNext(T value)`: Receives the next item in the stream
+ - `onError(Throwable throwable)`: Handles errors
+ - `onCompleted()`: Signals stream completion
+
+2. **Service Definition**: Protobuf service interface defining stream contracts
+
+3. **Protocol Handler**: Triple protocol manages connection lifecycle,
serialization, and resource cleanup
+
+---
+
+## Stream Types and Patterns
+
+### 1. Server Stream (One Request, Many Responses)
+
+Server streams are useful when a client sends a single request and the server
responds with multiple items over time.
+
+**Service Definition (Protobuf):**
+```protobuf
+syntax = "proto3";
+
+package org.apache.dubbo.demo;
+
+message Request {
+ string query = 1;
+}
+
+message Response {
+ string data = 1;
+}
+
+service DataService {
+ rpc streamData(Request) returns (stream Response);
+}
+```
+
+**Server Implementation:**
+```java
+import org.apache.dubbo.common.stream.StreamObserver;
+
+public class DataServiceImpl implements DataService {
+
+ @Override
+ public void streamData(Request request,
+ StreamObserver<Response> responseObserver) {
+ try {
+ // Fetch data in batches
+ List<String> dataItems = fetchData(request.getQuery());
+
+ for (String item : dataItems) {
+ Response response = Response.newBuilder()
+ .setData(item)
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ private List<String> fetchData(String query) {
+ // Implementation
+ return new ArrayList<>();
+ }
+}
+```
+
+**Client Consumption:**
+```java
+DataService.Stub stub = DataService.newStub(channel);
+
+StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processData(value.getData());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error: ", t);
+ // Handle error and potentially retry
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Stream completed successfully");
+ }
+ };
+
+Request request = Request.newBuilder()
+ .setQuery("important_data")
+ .build();
+
+stub.streamData(request, responseObserver);
+```
+
+### 2. Client Stream (Many Requests, One Response)
+
+Client streams allow multiple items to be sent to the server before receiving
a single response.
+
+**Service Definition:**
+```protobuf
+service BatchService {
+ rpc processBatch(stream BatchItem) returns (Result);
+}
+
+message BatchItem {
+ int64 id = 1;
+ bytes data = 2;
+}
+
+message Result {
+ int32 processedCount = 1;
+ string status = 2;
+}
+```
+
+**Server Implementation:**
+```java
+public class BatchServiceImpl implements BatchService {
+
+ @Override
+ public StreamObserver<BatchItem> processBatch(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<BatchItem>() {
+ private int count = 0;
+
+ @Override
+ public void onNext(BatchItem value) {
+ try {
+ processSingleItem(value);
+ count++;
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Client stream error: ", t);
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ Result result = Result.newBuilder()
+ .setProcessedCount(count)
+ .setStatus("SUCCESS")
+ .build();
+ responseObserver.onNext(result);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processSingleItem(BatchItem item) {
+ // Implementation
+ }
+}
+```
+
+**Client Implementation:**
+```java
+BatchService.Stub stub = BatchService.newStub(channel);
+
+StreamObserver<Result> responseObserver =
+ new StreamObserver<Result>() {
+ @Override
+ public void onNext(Result value) {
+ logger.info("Processed {} items with status: {}",
+ value.getProcessedCount(),
+ value.getStatus());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Batch processing error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Batch completed");
+ }
+ };
+
+StreamObserver<BatchItem> requestObserver =
+ stub.processBatch(responseObserver);
+
+for (BatchItem item : itemsToProcess) {
+ requestObserver.onNext(item);
+}
+
+requestObserver.onCompleted();
+```
+
+### 3. Bidirectional Stream (Many Requests, Many Responses)
+
+Bidirectional streams enable independent request and response flows, useful
for interactive applications.
+
+**Service Definition:**
+```protobuf
+service ChatService {
+ rpc chat(stream ChatMessage) returns (stream ChatMessage);
+}
+
+message ChatMessage {
+ string sender = 1;
+ string content = 2;
+ int64 timestamp = 3;
+}
+```
+
+**Server Implementation:**
+```java
+public class ChatServiceImpl implements ChatService {
+
+ @Override
+ public StreamObserver<ChatMessage> chat(
+ StreamObserver<ChatMessage> responseObserver) {
+
+ return new StreamObserver<ChatMessage>() {
+ private String clientId = UUID.randomUUID().toString();
+
+ @Override
+ public void onNext(ChatMessage message) {
+ // Broadcast to all connected clients
+ broadcastMessage(message, clientId);
+
+ // Echo back with server processing
+ ChatMessage response = ChatMessage.newBuilder()
+ .setSender("SERVER")
+ .setContent("Received: " + message.getContent())
+ .setTimestamp(System.currentTimeMillis())
+ .build();
+ responseObserver.onNext(response);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Chat error for {}: ", clientId, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ logger.info("Chat session completed for {}", clientId);
+ disconnectClient(clientId);
+ }
+ };
+ }
+
+ private void broadcastMessage(ChatMessage message, String senderId) {
+ // Broadcast logic
+ }
+
+ private void disconnectClient(String clientId) {
+ // Cleanup logic
+ }
+}
+```
+
+---
+
+## Handling Network Interruptions
+
+Network instability is a reality in production. Proper handling of
interruptions is critical for stream reliability.
+
+### 1. Connection Timeout Configuration
+
+Configure appropriate timeouts in your Dubbo application:
+
+**YAML Configuration:**
+```yaml
+dubbo:
+ consumer:
+ timeout: 30000 # 30 seconds for unary calls
+ provider:
+ timeout: 30000
+ protocol:
+ name: triple
+ port: 50051
+ # Stream-specific settings
+ parameters:
+ keep-alive-time: 30000 # Milliseconds
+ keep-alive-timeout: 10000
+ max-idle-time: 600000 # 10 minutes
+```
+
+**Java Configuration:**
+```java
+@Configuration
+public class DubboConfig {
+
+ @Bean
+ public ServiceConfig<DataService> dataServiceConfig() {
+ ServiceConfig<DataService> config = new ServiceConfig<>();
+ config.setInterface(DataService.class);
+ config.setRef(new DataServiceImpl());
+ config.setProtocol(tripleProtocolConfig());
+ return config;
+ }
+
+ @Bean
+ public ProtocolConfig tripleProtocolConfig() {
+ ProtocolConfig config = new ProtocolConfig();
+ config.setName("triple");
+ config.setPort(50051);
+ config.getParameters().put("keep-alive-time", "30000");
+ config.getParameters().put("keep-alive-timeout", "10000");
+ return config;
+ }
+}
+```
+
+### 2. Graceful Degradation
+
+Implement fallback mechanisms for stream failures:
+
+```java
+public class ResilientDataService {
+
+ private final DataService.Stub primaryStub;
+ private final DataService.Stub fallbackStub;
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ public void streamDataWithFallback(Request request) {
+ final AtomicInteger errorCount = new AtomicInteger(0);
+ final CountDownLatch completionLatch = new CountDownLatch(1);
+
+ StreamObserver<Response> responseObserver =
+ new StreamObserver<Response>() {
+
+ @Override
+ public void onNext(Response value) {
+ processResponse(value);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ errorCount.incrementAndGet();
+ logger.warn("Primary stream failed, attempting fallback",
t);
+
+ if (errorCount.get() >= 3) {
+ logger.error("Fallback exhausted");
+ completionLatch.countDown();
+ return;
+ }
+
+ // Use fallback service
+ primaryStub.streamData(request, this);
+ }
+
+ @Override
+ public void onCompleted() {
+ completionLatch.countDown();
+ }
+ };
+
+ primaryStub.streamData(request, responseObserver);
+
+ try {
+ completionLatch.await(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void processResponse(Response value) {
+ // Process response
+ }
+}
+```
+
+### 3. Heartbeat and Keepalive Mechanism
+
+Maintain stream connectivity with periodic heartbeats:
+
+```java
+public class HeartbeatStreamService {
+
+ @Override
+ public StreamObserver<StreamMessage> streamWithHeartbeat(
+ StreamObserver<StreamMessage> responseObserver) {
+
+ ScheduledExecutorService executor =
+ Executors.newScheduledThreadPool(1);
+
+ return new StreamObserver<StreamMessage>() {
+ private volatile long lastActivityTime =
+ System.currentTimeMillis();
+
+ @Override
+ public void onNext(StreamMessage message) {
+ lastActivityTime = System.currentTimeMillis();
+ processMessage(message);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ executor.shutdownNow();
+ }
+
+ @Override
+ public void onCompleted() {
+ executor.shutdownNow();
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void processMessage(StreamMessage message) {
+ // Implementation
+ }
+}
+```
+
+---
+
+## Strong Consistency Strategies
+
+Achieving strong consistency in streaming scenarios requires careful
coordination between client and server.
+
+### 1. Message Ordering Guarantees
+
+Ensure messages are processed in order:
+
+```java
+public class OrderedStreamProcessor {
+
+ private final AtomicLong lastProcessedSequence =
+ new AtomicLong(-1);
+
+ public void processOrderedStream(
+ StreamMessage message,
+ StreamObserver<StreamMessage> responseObserver) {
+
+ long currentSequence = message.getSequence();
+ long expectedSequence = lastProcessedSequence.get() + 1;
+
+ if (currentSequence == expectedSequence) {
+ // Process message
+ processMessage(message);
+ lastProcessedSequence.set(currentSequence);
+
+ // Send acknowledgment
+ StreamMessage ack = StreamMessage.newBuilder()
+ .setSequence(currentSequence)
+ .setType("ACK")
+ .build();
+ responseObserver.onNext(ack);
+
+ } else if (currentSequence > expectedSequence) {
+ // Out of order - request retransmission
+ StreamMessage nack = StreamMessage.newBuilder()
+ .setSequence(expectedSequence)
+ .setType("NACK")
+ .build();
+ responseObserver.onNext(nack);
+
+ } else {
+ // Duplicate - send ACK but don't process
+ StreamMessage ack = StreamMessage.newBuilder()
+ .setSequence(currentSequence)
+ .setType("ACK")
+ .build();
+ responseObserver.onNext(ack);
+ }
+ }
+
+ private void processMessage(StreamMessage message) {
+ // Implementation
+ }
+}
+```
+
+### 2. Transactional Stream Processing
+
+Batch messages and process them transactionally:
+
+```java
+public class TransactionalStreamService {
+
+ private final int batchSize = 100;
+ private final List<StreamMessage> batch = new ArrayList<>();
+ private final Object lock = new Object();
+
+ @Override
+ public StreamObserver<StreamMessage> processTransactional(
+ StreamObserver<Result> responseObserver) {
+
+ return new StreamObserver<StreamMessage>() {
+
+ @Override
+ public void onNext(StreamMessage message) {
+ synchronized (lock) {
+ batch.add(message);
+
+ if (batch.size() >= batchSize) {
+ processBatch();
+ }
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ synchronized (lock) {
+ // Rollback uncommitted batch
+ batch.clear();
+ }
+ responseObserver.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ synchronized (lock) {
+ if (!batch.isEmpty()) {
+ processBatch();
+ }
+ }
+ responseObserver.onCompleted();
+ }
+
+ private void processBatch() {
+ try {
+ // Execute in transaction
+ executeTransaction(batch);
+
+ Result result = Result.newBuilder()
+ .setSuccess(true)
+ .setCount(batch.size())
+ .build();
+ responseObserver.onNext(result);
+
+ batch.clear();
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+ };
+ }
+
+ private void executeTransaction(List<StreamMessage> messages) {
+ // Transaction implementation
+ }
+}
+```
+
+---
+
+## Idempotency Implementation
+
+Idempotency ensures that duplicate messages (often due to retries) don't cause
unintended side effects.
+
+### 1. Message ID Tracking
+
+Use message IDs to detect and handle duplicates:
+
+```java
+public class IdempotentStreamProcessor {
+
+ private final Set<String> processedMessageIds =
+ Collections.synchronizedSet(new HashSet<>());
+
+ @Override
+ public StreamObserver<Message> idempotentStream(
+ StreamObserver<Response> responseObserver) {
+
+ return new StreamObserver<Message>() {
+
+ @Override
+ public void onNext(Message message) {
+ String messageId = message.getId();
+
+ if (processedMessageIds.contains(messageId)) {
+ // Duplicate - send previous response
+ Response response = getCachedResponse(messageId);
+ responseObserver.onNext(response);
+ return;
+ }
+
+ // Process message
+ try {
+ Response response = processAndCacheResult(message);
+ processedMessageIds.add(messageId);
+ responseObserver.onNext(response);
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private Response processAndCacheResult(Message message) {
+ // Implementation
+ return Response.newBuilder().build();
+ }
+
+ private Response getCachedResponse(String messageId) {
+ // Retrieve from cache
+ return Response.newBuilder().build();
+ }
+}
+```
+
+### 2. Idempotency Key Storage
+
+Persist idempotency keys for longer-term deduplication:
+
+```java
+public class PersistentIdempotencyService {
+
+ private final IdempotencyKeyRepository repository;
+
+ @Override
+ public StreamObserver<Operation> persistentIdempotentOps(
+ StreamObserver<OperationResult> responseObserver) {
+
+ return new StreamObserver<Operation>() {
+
+ @Override
+ public void onNext(Operation op) {
+ String idempotencyKey = op.getIdempotencyKey();
+
+ // Check database for existing result
+ Optional<OperationResult> cachedResult =
+ repository.findByIdempotencyKey(idempotencyKey);
+
+ if (cachedResult.isPresent()) {
+ responseObserver.onNext(cachedResult.get());
+ return;
+ }
+
+ try {
+ OperationResult result = executeOperation(op);
+
+ // Store result with idempotency key
+ repository.save(new IdempotencyRecord(
+ idempotencyKey,
+ result,
+ System.currentTimeMillis()
+ ));
+
+ responseObserver.onNext(result);
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Operation stream error: ", t);
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private OperationResult executeOperation(Operation op) {
+ // Implementation
+ return OperationResult.newBuilder().build();
+ }
+}
+```
+
+---
+
+## Checkpoint Mechanisms
+
+Checkpoints enable resuming streams from the last known good state after
interruptions.
+
+### 1. Periodic Checkpoint Recording
+
+Record progress at regular intervals:
+
+```java
+public class CheckpointedStreamProcessor {
+
+ private final CheckpointRepository repository;
+ private volatile long lastCheckpoint = 0;
+ private final long checkpointInterval = 1000; // 1 second
+
+ @Override
+ public StreamObserver<DataChunk> streamWithCheckpoints(
+ String streamId,
+ StreamObserver<CheckpointAck> responseObserver) {
+
+ return new StreamObserver<DataChunk>() {
+ private long processedBytes = 0;
+ private long lastCheckpointTime = System.currentTimeMillis();
+
+ @Override
+ public void onNext(DataChunk chunk) {
+ try {
+ processChunk(chunk);
+ processedBytes += chunk.getData().size();
+
+ // Record checkpoint periodically
+ long now = System.currentTimeMillis();
+ if (now - lastCheckpointTime >= checkpointInterval) {
+ recordCheckpoint(streamId, processedBytes);
+
+ CheckpointAck ack = CheckpointAck.newBuilder()
+ .setStreamId(streamId)
+ .setCheckpointOffset(processedBytes)
+ .setTimestamp(now)
+ .build();
+ responseObserver.onNext(ack);
+
+ lastCheckpointTime = now;
+ }
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ logger.error("Stream error at byte offset {}: ",
+ processedBytes, t);
+ }
+
+ @Override
+ public void onCompleted() {
+ recordCheckpoint(streamId, processedBytes);
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private void recordCheckpoint(String streamId, long offset) {
+ repository.saveCheckpoint(
+ new Checkpoint(streamId, offset, System.currentTimeMillis())
+ );
+ }
+
+ private void processChunk(DataChunk chunk) {
+ // Implementation
+ }
+}
+```
+
+### 2. Stream Resume from Checkpoint
+
+Resume streaming from last checkpoint:
+
+```java
+public class CheckpointResumeService {
+
+ private final CheckpointRepository repository;
+
+ public void resumeStream(String streamId,
+ StreamObserver<DataChunk> responseObserver) {
+
+ // Retrieve last checkpoint
+ Optional<Checkpoint> lastCheckpoint =
+ repository.getLatestCheckpoint(streamId);
+
+ long resumeFromOffset = lastCheckpoint
+ .map(Checkpoint::getOffset)
+ .orElse(0L);
+
+ logger.info("Resuming stream {} from offset {}",
+ streamId, resumeFromOffset);
+
+ try {
+ // Stream data starting from checkpoint
+ streamDataFromOffset(streamId, resumeFromOffset,
+ responseObserver);
+ } catch (Exception e) {
+ responseObserver.onError(e);
+ }
+ }
+
+ private void streamDataFromOffset(String streamId,
+ long offset,
+ StreamObserver<DataChunk> observer) {
+ // Implementation: fetch and stream data from offset
+ }
+}
+```
+
+---
+
+## Error Handling and Recovery
+
+Robust error handling is essential for production stream services.
+
+### 1. Comprehensive Error Categorization
+
+Differentiate between recoverable and fatal errors:
+
+```java
+public class ErrorHandlingStreamService {
+
+ public enum ErrorType {
+ TRANSIENT, // Retry recommended
+ PERMANENT, // Don't retry
+ DEGRADED // Continue with reduced functionality
+ }
+
+ @Override
+ public StreamObserver<Request> robustStream(
+ StreamObserver<Response> responseObserver) {
+
+ return new StreamObserver<Request>() {
+ private int retryCount = 0;
+ private static final int MAX_RETRIES = 3;
+
+ @Override
+ public void onNext(Request request) {
+ try {
+ processRequest(request, responseObserver);
+ retryCount = 0; // Reset on success
+ } catch (Exception e) {
+ handleStreamError(e, responseObserver);
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ ErrorType errorType = classifyError(t);
+
+ switch (errorType) {
+ case TRANSIENT:
+ if (retryCount < MAX_RETRIES) {
+ retryCount++;
+ logger.info("Transient error, retry {} of {}",
+ retryCount, MAX_RETRIES);
+ // Retry logic
+ } else {
+ logger.error("Max retries exceeded");
+ responseObserver.onError(t);
+ }
+ break;
+
+ case PERMANENT:
+ logger.error("Permanent error, aborting stream", t);
+ responseObserver.onError(t);
+ break;
+
+ case DEGRADED:
+ logger.warn("Degraded mode", t);
+ // Continue with degraded service
+ break;
+ }
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+
+ private void processRequest(Request request,
+ StreamObserver<Response> observer) {
+ // Implementation
+ }
+
+ private void handleStreamError(Exception e,
+ StreamObserver<Response> observer) {
+ // Implementation
+ }
+
+ private ErrorType classifyError(Throwable t) {
+ if (t instanceof TimeoutException ||
+ t instanceof SocketTimeoutException) {
+ return ErrorType.TRANSIENT;
+ } else if (t instanceof IllegalArgumentException) {
+ return ErrorType.PERMANENT;
+ }
+ return ErrorType.DEGRADED;
+ }
+ };
+ }
+}
+```
+
+### 2. Circuit Breaker Pattern
+
+Implement circuit breaker for failing services:
+
+```java
+public class CircuitBreakerStreamService {
+
+ public enum CircuitState {
+ CLOSED, // Normal operation
+ OPEN, // Fail fast
+ HALF_OPEN // Testing recovery
+ }
+
+ private volatile CircuitState state = CircuitState.CLOSED;
+ private volatile long lastFailureTime = 0;
+ private volatile int failureCount = 0;
+ private static final int FAILURE_THRESHOLD = 5;
+ private static final long RESET_TIMEOUT = 30000; // 30 seconds
+
+ public StreamObserver<Request> streamWithCircuitBreaker(
+ StreamObserver<Response> responseObserver) {
+
+ return new StreamObserver<Request>() {
+
+ @Override
+ public void onNext(Request request) {
+ if (state == CircuitState.OPEN) {
+ if (System.currentTimeMillis() - lastFailureTime
+ > RESET_TIMEOUT) {
+ state = CircuitState.HALF_OPEN;
+ logger.info("Circuit breaker: HALF_OPEN");
+ } else {
+ responseObserver.onError(
+ new RuntimeException("Circuit breaker OPEN"));
+ return;
+ }
+ }
+
+ try {
+ processRequest(request, responseObserver);
+
+ if (state == CircuitState.HALF_OPEN) {
+ state = CircuitState.CLOSED;
+ failureCount = 0;
+ logger.info("Circuit breaker: CLOSED");
+ }
+ } catch (Exception e) {
+ failureCount++;
+ lastFailureTime = System.currentTimeMillis();
+
+ if (failureCount >= FAILURE_THRESHOLD) {
+ state = CircuitState.OPEN;
+ logger.warn("Circuit breaker: OPEN");
+ }
+
+ responseObserver.onError(e);
+ }
Review Comment:
The circuit breaker implementation has race conditions. While individual
fields are marked `volatile`, the compound operations (check-then-act patterns)
are not atomic. For example, lines 992-995 read the state and time, then update
state, but another thread could modify state in between. Lines 1012-1018
increment failureCount and check/update state non-atomically. Multiple
concurrent threads could incorrectly increment the failure count or change
states. Consider using AtomicInteger for failureCount and synchronizing the
state transitions, or use a proper circuit breaker library like Resilience4j.
--
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]
