[ISSUE #8102]🐛Preserve ACL permission denied semantics

rocketmq-auth/src/authorization/provider.rs

@@ -111,9 +111,7 @@ impl From<AuthorizationError> for RocketMQError {
     fn from(error: AuthorizationError) -> Self {
         let message = error.to_string();
         match error {
-            AuthorizationError::PermissionDenied { .. } => {
-                RocketMQError::Authentication(AuthError::AuthorizationFailed(message))
-            }
+            AuthorizationError::PermissionDenied { .. } => RocketMQError::BrokerPermissionDenied { operation: message },
             AuthorizationError::SubjectNotFound(_)
             | AuthorizationError::ResourceNotFound(_)
             | AuthorizationError::InvalidContext(_) => RocketMQError::illegal_argument(message),
@@ -693,10 +691,7 @@ mod tests {
             resource: "topic:test".to_string(),
             reason: "insufficient permissions".to_string(),
         });
-        assert!(matches!(
-            denied,
-            RocketMQError::Authentication(AuthError::AuthorizationFailed(_))
-        ));
+        assert!(matches!(denied, RocketMQError::BrokerPermissionDenied { .. }));
 
         let invalid = RocketMQError::from(AuthorizationError::InvalidContext("missing subject".to_string()));
         assert!(matches!(invalid, RocketMQError::IllegalArgument(_)));

rocketmq-auth/src/lib.rs

@@ -142,9 +142,15 @@ pub mod bench_support {
 
         let deadline = Instant::now() + Duration::from_secs(2);
         let reload_success = loop {
-            let user = authn_provider.get_user("alice").await?;
-            if user.password().map(|value| value.as_str()) == Some("second") {
-                break true;
+            match authn_provider.get_user("alice").await {
+                Ok(user) if user.password().map(|value| value.as_str()) == Some("second") => {
+                    break true;
+                }
+                Ok(_) => {}
+                Err(rocketmq_error::RocketMQError::Authentication(rocketmq_error::AuthError::UserNotFound(
+                    username,
+                ))) if username == "alice" => {}
+                Err(error) => return Err(error),
             }
             if Instant::now() >= deadline {
                 break false;
@@ -294,21 +300,46 @@ pub mod bench_support {
     }
 
     fn write_acl_file(path: &std::path::Path, secret: &str) -> RocketMQResult<()> {
-        fs::write(
-            path,
-            format!(
-                r#"
+        let content = format!(
+            r#"
 accounts:
   - accessKey: alice
     secretKey: {secret}
 "#
-            ),
-        )
-        .map_err(|error| {
+        );
+        let temp_file = temp_acl_file_path(path);
+        fs::write(&temp_file, content).map_err(|error| {
+            rocketmq_error::RocketMQError::storage_write_failed(temp_file.display().to_string(), error.to_string())
+        })?;
+
+        replace_acl_file(&temp_file, path)
+    }
+
+    fn temp_acl_file_path(path: &std::path::Path) -> PathBuf {
+        let file_name = path
+            .file_name()
+            .and_then(|value| value.to_str())
+            .unwrap_or("plain_acl.yml");
+        let id = NEXT_ACL_WATCHER_PROBE_ID.fetch_add(1, Ordering::Relaxed);
+        path.with_file_name(format!(".{file_name}.{id}.tmp"))
+    }
+
+    #[cfg(not(windows))]
+    fn replace_acl_file(temp_file: &std::path::Path, path: &std::path::Path) -> RocketMQResult<()> {
+        fs::rename(temp_file, path).map_err(|error| {
             rocketmq_error::RocketMQError::storage_write_failed(path.display().to_string(), error.to_string())
         })
     }
 
+    #[cfg(windows)]
+    fn replace_acl_file(temp_file: &std::path::Path, path: &std::path::Path) -> RocketMQResult<()> {
+        fs::copy(temp_file, path).map_err(|error| {
+            rocketmq_error::RocketMQError::storage_write_failed(path.display().to_string(), error.to_string())
+        })?;
+        let _ = fs::remove_file(temp_file);
+        Ok(())
+    }
+
     impl From<crate::runtime_bridge::AuthSyncBridgeSnapshot> for AuthSyncBridgeCounterSnapshot {
         fn from(snapshot: crate::runtime_bridge::AuthSyncBridgeSnapshot) -> Self {
             Self {

rocketmq-observability/src/init.rs

@@ -201,7 +201,10 @@ pub(crate) fn init_telemetry_providers(config: &ObservabilityConfig) -> Result<T
         return Ok(TelemetryGuard::noop());
     }
 
+    #[cfg(any(feature = "otel-metrics", feature = "otel-traces", feature = "otel-logs"))]
     let mut guard = TelemetryGuard::noop();
+    #[cfg(not(any(feature = "otel-metrics", feature = "otel-traces", feature = "otel-logs")))]
+    let guard = TelemetryGuard::noop();
 
     if config.metrics.enabled {
         #[cfg(feature = "otel-metrics")]

rocketmq-observability/src/logging.rs

@@ -148,7 +148,7 @@ fn init_log_tracer() {
 
 fn build_subscriber_layers(
     config: &TelemetryBootstrapConfig,
-    telemetry_guard: &TelemetryGuard,
+    _telemetry_guard: &TelemetryGuard,
 ) -> Result<(Vec<BoxedRegistryLayer>, LoggingGuard), ObservabilityError> {
     let mut layers = Vec::new();
     let mut logging_guard = LoggingGuard::noop();
@@ -168,15 +168,15 @@ fn build_subscriber_layers(
     }
 
     #[cfg(feature = "otel-traces")]
-    if let Some(tracer_provider) = telemetry_guard.tracer_provider() {
+    if let Some(tracer_provider) = _telemetry_guard.tracer_provider() {
         layers.push(Box::new(crate::trace::build_tracing_layer(
             &config.observability,
             tracer_provider,
         )));
     }
 
     #[cfg(feature = "otel-logs")]
-    if let Some(logger_provider) = telemetry_guard.logger_provider() {
+    if let Some(logger_provider) = _telemetry_guard.logger_provider() {
         layers.push(Box::new(crate::logs::bridge::build_logs_layer(logger_provider)));
     }