Merge branch 'main' into yiming/non-async-complete-barrier

.github/dependabot.yml

@@ -26,6 +26,7 @@ updates:
         - "prost*"
     opentelemetry:
       patterns:
+        - "opentelemetry"
         - "opentelemetry*"
         - "tracing-opentelemetry"
     mysql:

Cargo.toml

@@ -153,7 +153,12 @@ tikv-jemallocator = { git = "https://github.com/risingwavelabs/jemallocator.git"
   "profiling",
   "stats",
 ], rev = "64a2d9" }
+opentelemetry = "0.21"
+opentelemetry-otlp = "0.14"
+opentelemetry_sdk = { version = "0.21", default-features = false }
+opentelemetry-semantic-conventions = "0.13"
 tokio-util = "0.7"
+tracing-opentelemetry = "0.22"
 
 risingwave_backup = { path = "./src/storage/backup" }
 risingwave_batch = { path = "./src/batch" }

proto/stream_service.proto

@@ -46,6 +46,7 @@ message DropActorsResponse {
 
 message ForceStopActorsRequest {
   string request_id = 1;
+  uint64 prev_epoch = 2;
 }
 
 message ForceStopActorsResponse {

src/bench/Cargo.toml

@@ -23,7 +23,7 @@ futures-async-stream = { workspace = true }
 hdrhistogram = "7"
 itertools = "0.12"
 libc = "0.2"
-opentelemetry = { version = "0.20", default-features = false, features = ["rt-tokio"], optional = true }
+opentelemetry = { workspace = true, optional = true }
 parking_lot = "0.12"
 prometheus = { version = "0.13", features = ["process"] }
 rand = "0.8"

src/common/Cargo.toml

@@ -61,7 +61,8 @@ memcomparable = { version = "0.2", features = ["decimal"] }
 num-integer = "0.1"
 num-traits = "0.2"
 number_prefix = "0.4.0"
-opentelemetry = { version = "0.20", default-features = false }
+opentelemetry = { workspace = true }
+opentelemetry_sdk = { workspace = true }
 parking_lot = "0.12"
 parse-display = "0.9"
 paste = "1"
@@ -110,7 +111,7 @@ toml = "0.8"
 tonic = { workspace = true }
 tracing = "0.1"
 tracing-futures = { version = "0.2", features = ["futures-03"] }
-tracing-opentelemetry = "0.21"
+tracing-opentelemetry = { workspace = true }
 tracing-subscriber = "0.3.17"
 twox-hash = "1"
 url = "2"

src/common/src/util/tracing.rs

@@ -19,7 +19,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use opentelemetry::propagation::TextMapPropagator;
-use opentelemetry::sdk::propagation::TraceContextPropagator;
+use opentelemetry_sdk::propagation::TraceContextPropagator;
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 
 /// Context for tracing used for propagating tracing information in a distributed system.

src/compute/src/rpc/service/stream_service.rs

@@ -116,7 +116,7 @@ impl StreamService for StreamServiceImpl {
         request: Request<ForceStopActorsRequest>,
     ) -> std::result::Result<Response<ForceStopActorsResponse>, Status> {
         let req = request.into_inner();
-        self.mgr.reset().await;
+        self.mgr.reset(req.prev_epoch).await;
         Ok(Response::new(ForceStopActorsResponse {
             request_id: req.request_id,
             status: None,