fix(telemtry): remove redundant report_event during recovery

src/common/src/telemetry/mod.rs

@@ -20,8 +20,8 @@ use std::env;
 
 use risingwave_pb::telemetry::PbTelemetryClusterType;
 pub use risingwave_telemetry_event::{
-    current_timestamp, post_telemetry_report_pb, report_event_common, request_to_telemetry_event,
-    TelemetryError, TelemetryResult,
+    current_timestamp, post_telemetry_report_pb, report_event_common, TelemetryError,
+    TelemetryResult,
 };
 use serde::{Deserialize, Serialize};
 use sysinfo::System;

src/common/telemetry_event/src/lib.rs

@@ -51,6 +51,12 @@ pub fn report_event_common(
     attributes: Option<jsonbb::Value>, // any json string
     node: String,
 ) {
+    // REMOVE ME: To find out all report_event calls
+    println!(
+        "report_event_common backtrace:\n{}",
+        std::backtrace::Backtrace::force_capture()
+    );
+
     let event_tracking_id: String;
     if let Some(tracking_id) = TELEMETRY_TRACKING_ID.get() {
         event_tracking_id = tracking_id.to_string();
@@ -72,7 +78,7 @@ pub fn report_event_common(
     );
 }
 
-pub fn request_to_telemetry_event(
+fn request_to_telemetry_event(
     tracking_id: String,
     event_stage: PbTelemetryEventStage,
     event_name: &str,

src/stream/src/from_proto/sink.rs

@@ -41,32 +41,6 @@ use crate::telemetry::report_event;
 
 pub struct SinkExecutorBuilder;
 
-fn telemetry_sink_build(
-    sink_id: &SinkId,
-    connector_name: &str,
-    sink_format_desc: &Option<SinkFormatDesc>,
-) {
-    let attr = sink_format_desc.as_ref().map(|f| {
-        let mut builder = jsonbb::Builder::<Vec<u8>>::new();
-        builder.begin_object();
-        builder.add_string("format");
-        builder.add_value(jsonbb::ValueRef::String(f.format.to_string().as_str()));
-        builder.add_string("encode");
-        builder.add_value(jsonbb::ValueRef::String(f.encode.to_string().as_str()));
-        builder.end_object();
-        builder.finish()
-    });
-
-    report_event(
-        PbTelemetryEventStage::CreateStreamJob,
-        "sink",
-        sink_id.sink_id() as i64,
-        Some(connector_name.to_owned()),
-        Some(PbTelemetryDatabaseObject::Sink),
-        attr,
-    )
-}
-
 fn resolve_pk_info(
     input_schema: &Schema,
     log_store_table: &Table,
@@ -259,8 +233,6 @@ impl ExecutorBuilder for SinkExecutorBuilder {
             connector, sink_id.sink_id, params.executor_id
         );
 
-        telemetry_sink_build(&sink_id, connector, &sink_param.format_desc);
-
         let exec = match node.log_store_type() {
             // Default value is the normal in memory log store to be backward compatible with the
             // previously unset value

src/stream/src/from_proto/source/mod.rs

@@ -22,41 +22,12 @@ pub use fs_fetch::FsFetchExecutorBuilder;
 use risingwave_common::catalog::TableId;
 use risingwave_connector::source::UPSTREAM_SOURCE_KEY;
 use risingwave_pb::catalog::PbStreamSourceInfo;
-use risingwave_pb::telemetry::{PbTelemetryDatabaseObject, PbTelemetryEventStage};
 
 use super::*;
-use crate::telemetry::report_event;
 
 fn get_connector_name(with_props: &BTreeMap<String, String>) -> String {
     with_props
         .get(UPSTREAM_SOURCE_KEY)
         .map(|s| s.to_lowercase())
         .unwrap_or_default()
 }
-
-fn telemetry_source_build(
-    source_type: &str, // "source" or "source backfill"
-    source_id: &TableId,
-    source_info: &PbStreamSourceInfo,
-    with_props: &BTreeMap<String, String>,
-) {
-    let mut builder = jsonbb::Builder::<Vec<u8>>::new();
-    builder.begin_object();
-    builder.add_string("format");
-    builder.add_value(jsonbb::ValueRef::String(source_info.format().as_str_name()));
-    builder.add_string("encode");
-    builder.add_value(jsonbb::ValueRef::String(
-        source_info.row_encode().as_str_name(),
-    ));
-    builder.end_object();
-    let value = builder.finish();
-
-    report_event(
-        PbTelemetryEventStage::CreateStreamJob,
-        source_type,
-        source_id.table_id as i64,
-        Some(get_connector_name(with_props)),
-        Some(PbTelemetryDatabaseObject::Source),
-        Some(value),
-    )
-}

src/stream/src/from_proto/source/trad_source.rs

@@ -110,8 +110,6 @@ pub fn create_source_desc_builder(
         });
     }
 
-    telemetry_source_build(source_type, source_id, &source_info, &with_properties);
-
     SourceDescBuilder::new(
         source_columns.clone(),
         params.env.source_metrics(),