From 5b05867c2b30aba46e40e002c8bd65689517d7fe Mon Sep 17 00:00:00 2001
From: "xiang.zhou" <xiang.zhou@emqx.io>
Date: Mon, 9 Jun 2025 10:54:45 +0800
Subject: [PATCH] adapter:check state when do with trans data

---
 include/neuron/otel/otel_manager.h |  2 ++
 src/adapter/adapter.c              | 15 +++++++++++++--
 src/adapter/driver/driver.c        | 14 --------------
 src/otel/otel_manager.c            | 10 ++++++++++
 4 files changed, 25 insertions(+), 16 deletions(-)

diff --git a/include/neuron/otel/otel_manager.h b/include/neuron/otel/otel_manager.h
index 25aad8a0..074d6657 100644
--- a/include/neuron/otel/otel_manager.h
+++ b/include/neuron/otel/otel_manager.h
@@ -42,6 +42,8 @@ neu_otel_trace_ctx neu_otel_find_trace2(void *req_ctx);
 void neu_otel_trace_set_final(neu_otel_trace_ctx ctx);
 
 void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num);
+void neu_otel_trace_reduce_expected_span_num(neu_otel_trace_ctx ctx,
+                                             uint32_t           num);
 uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx);
 
 neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id);
diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c
index afc1d142..8189f597 100644
--- a/src/adapter/adapter.c
+++ b/src/adapter/adapter.c
@@ -123,8 +123,19 @@ static void *adapter_consumer(void *arg)
         nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
                    adapter->name, header->sender, header->ctx,
                    neu_reqresp_type_string(header->type), n);
-        adapter->module->intf_funs->request(
-            adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+        if (adapter->state == NEU_NODE_RUNNING_STATE_RUNNING) {
+            adapter->module->intf_funs->request(
+                adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+        } else {
+            void *ctx = ((neu_reqresp_trans_data_t *) &header[1])->trace_ctx;
+            if (neu_otel_data_is_started() && ctx) {
+                neu_otel_trace_ctx trace = neu_otel_find_trace(ctx);
+                if (trace) {
+                    neu_otel_trace_reduce_expected_span_num(trace, 1);
+                }
+            }
+        }
+
         neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
         neu_msg_free(msg);
     }
diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c
index d95a04ff..b4344448 100644
--- a/src/adapter/driver/driver.c
+++ b/src/adapter/driver/driver.c
@@ -2289,20 +2289,6 @@ static int report_callback(void *usr_data)
 
             utarray_foreach(group->apps, sub_app_t *, app)
             {
-
-                neu_adapter_t *sub_app =
-                    neu_node_manager_find(g_manager->node_manager, app->app);
-                if (sub_app != NULL) {
-                    if (sub_app->state != NEU_NODE_RUNNING_STATE_RUNNING) {
-                        neu_trans_data_free(data);
-                        continue;
-                    }
-                } else {
-                    nlog_error("sub app %s not exist", app->app);
-                    neu_trans_data_free(data);
-                    continue;
-                }
-
                 if (group->driver->adapter.cb_funs.responseto(
                         &group->driver->adapter, &header, data, app->addr) !=
                     0) {
diff --git a/src/otel/otel_manager.c b/src/otel/otel_manager.c
index dfff5fd2..4f19b02c 100644
--- a/src/otel/otel_manager.c
+++ b/src/otel/otel_manager.c
@@ -540,6 +540,16 @@ void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num)
     trace_ctx->expected_span_num = num;
 }
 
+void neu_otel_trace_reduce_expected_span_num(neu_otel_trace_ctx ctx,
+                                             uint32_t           num)
+{
+    trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;
+    trace_ctx->expected_span_num -= num;
+    if (trace_ctx->expected_span_num <= 0) {
+        trace_ctx->final = true;
+    }
+}
+
 uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx)
 {
     trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;