diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 10d189e..832700c 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -91,6 +91,7 @@
           - ""
           - "--features management"
           - "--features vendored"
+          - "--features kafka-reporter"
           - "--all-features"
     runs-on: ubuntu-20.04
     steps:
diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml
index d1fab52..79609e2 100644
--- a/.github/workflows/e2e.yml
+++ b/.github/workflows/e2e.yml
@@ -40,3 +40,5 @@
           pip3 install setuptools
           pip3 install -r requirements.txt
           python3 e2e/run_e2e.py --expected_file=e2e/data/expected_context.yaml --max_retry_times=3 --target_path=/ping
+          docker-compose -f docker-compose.e2e.yml logs producer consumer
+          docker-compose -f docker-compose.e2e.yml exec -T producer /build/target/release/e2e-kafka
diff --git a/Cargo.toml b/Cargo.toml
index da1df32..909adb7 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -36,6 +36,7 @@
 [features]
 management = ["hostname", "systemstat"]
 vendored = ["protobuf-src"]
+kafka-reporter = ["rdkafka"]
 
 mock = []  # For internal integration testing only, do not use.
 
@@ -46,11 +47,13 @@
 futures-core = "0.3.21"
 futures-util = "0.3.21"
 hostname = { version = "0.3.1", optional = true }
+libz-sys = "1.1.9"
 once_cell = "1.14.0"
 parking_lot = "0.12.1"
 portable-atomic = { version = "0.3.13", features = ["float"] }
 prost = "0.11.0"
 prost-derive = "0.11.0"
+rdkafka = { version = "0.32.2", optional = true }
 serde = { version = "1.0.143", features = ["derive"] }
 systemstat = { version = "0.2.0", optional = true }
 thiserror = "1.0.32"
diff --git a/README.md b/README.md
index 715001e..d01ab4a 100644
--- a/README.md
+++ b/README.md
@@ -212,6 +212,42 @@
 }
 ```
 
+# Advanced Reporter
+
+The advanced report provides an alternative way to submit the agent collected data to the backend.
+
+## kafka reporter
+
+The Kafka reporter plugin support report traces, metrics, logs, instance properties to Kafka cluster.
+
+Make sure the feature `kafka-reporter` is enabled.
+
+```rust
+#[cfg(feature = "kafka-reporter")]
+mod example {
+    use skywalking::reporter::Report;
+    use skywalking::reporter::kafka::{KafkaReportBuilder, KafkaReporter, RDKafkaClientConfig};
+
+    async fn do_something(reporter: &impl Report) {
+        // ....
+    }
+
+    async fn foo() {
+        let mut client_config = RDKafkaClientConfig::new();
+        client_config
+            .set("bootstrap.servers", "broker:9092")
+            .set("message.timeout.ms", "6000");
+
+        let (reporter, reporting) = KafkaReportBuilder::new(client_config).build().await.unwrap();
+        let handle = reporting.spawn();
+
+        do_something(&reporter);
+
+        handle.await.unwrap();
+    }
+}
+```
+
 # How to compile?
 
 If you have `skywalking-(VERSION).crate`, you can unpack it with the way as follows:
diff --git a/dist-material/LICENSE b/dist-material/LICENSE
index 6c5eda5..a7c1296 100644
--- a/dist-material/LICENSE
+++ b/dist-material/LICENSE
@@ -220,9 +220,11 @@
     https://crates.io/crates/autocfg/1.1.0 1.1.0 Apache-2.0 OR MIT
     https://crates.io/crates/base64/0.13.0 0.13.0 Apache-2.0 OR MIT
     https://crates.io/crates/bitflags/1.3.1 1.3.1 Apache-2.0 OR MIT
-    https://crates.io/crates/cc/1.0.0 1.0.0 Apache-2.0 OR MIT
+    https://crates.io/crates/cc/1.0.18 1.0.18 Apache-2.0 OR MIT
     https://crates.io/crates/cfg-if/0.1.2 0.1.2 Apache-2.0 OR MIT
     https://crates.io/crates/cfg-if/1.0.0 1.0.0 Apache-2.0 OR MIT
+    https://crates.io/crates/derivative/2.0.0 2.0.0 Apache-2.0 OR MIT
+    https://crates.io/crates/dtoa/0.4.0 0.4.0 Apache-2.0 OR MIT
     https://crates.io/crates/either/1.0.0 1.0.0 Apache-2.0 OR MIT
     https://crates.io/crates/fixedbitset/0.4.0 0.4.0 Apache-2.0 OR MIT
     https://crates.io/crates/fnv/1.0.5 1.0.5 Apache-2.0 OR MIT
@@ -242,10 +244,12 @@
     https://crates.io/crates/hyper-timeout/0.4.0 0.4.0 Apache-2.0 OR MIT
     https://crates.io/crates/indexmap/1.6.2 1.6.2 Apache-2.0 OR MIT
     https://crates.io/crates/itertools/0.10.0 0.10.0 Apache-2.0 OR MIT
+    https://crates.io/crates/itoa/0.3.0 0.3.0 Apache-2.0 OR MIT
     https://crates.io/crates/itoa/0.4.1 0.4.1 Apache-2.0 OR MIT
     https://crates.io/crates/itoa/1.0.1 1.0.1 Apache-2.0 OR MIT
     https://crates.io/crates/lazy_static/1.4.0 1.4.0 Apache-2.0 OR MIT
     https://crates.io/crates/libc/0.2.114 0.2.114 Apache-2.0 OR MIT
+    https://crates.io/crates/libz-sys/1.1.9 1.1.9 Apache-2.0 OR MIT
     https://crates.io/crates/lock_api/0.4.6 0.4.6 Apache-2.0 OR MIT
     https://crates.io/crates/log/0.4.17 0.4.17 Apache-2.0 OR MIT
     https://crates.io/crates/match_cfg/0.1.0 0.1.0 Apache-2.0 OR MIT
@@ -254,6 +258,7 @@
     https://crates.io/crates/miow/0.3.6 0.3.6 Apache-2.0 OR MIT
     https://crates.io/crates/multimap/0.8.0 0.8.0 Apache-2.0 OR MIT
     https://crates.io/crates/ntapi/0.3.0 0.3.0 Apache-2.0 OR MIT
+    https://crates.io/crates/num-traits/0.1.32 0.1.32 Apache-2.0 OR MIT
     https://crates.io/crates/num_cpus/1.8.0 1.8.0 Apache-2.0 OR MIT
     https://crates.io/crates/num_threads/0.1.2 0.1.2 Apache-2.0 OR MIT
     https://crates.io/crates/once_cell/1.14.0 1.14.0 Apache-2.0 OR MIT
@@ -267,9 +272,11 @@
     https://crates.io/crates/pin-project-internal/1.0.2 1.0.2 Apache-2.0 OR MIT
     https://crates.io/crates/pin-project-lite/0.2.9 0.2.9 Apache-2.0 OR MIT
     https://crates.io/crates/pin-utils/0.1.0 0.1.0 Apache-2.0 OR MIT
+    https://crates.io/crates/pkg-config/0.3.9 0.3.9 Apache-2.0 OR MIT
     https://crates.io/crates/portable-atomic/0.3.13 0.3.13 Apache-2.0 OR MIT
     https://crates.io/crates/ppv-lite86/0.2.8 0.2.8 Apache-2.0 OR MIT
     https://crates.io/crates/prettyplease/0.1.0 0.1.0 Apache-2.0 OR MIT
+    https://crates.io/crates/proc-macro-crate/0.1.4 0.1.4 Apache-2.0 OR MIT
     https://crates.io/crates/proc-macro2/1.0.32 1.0.32 Apache-2.0 OR MIT
     https://crates.io/crates/quote/1.0.0 1.0.0 Apache-2.0 OR MIT
     https://crates.io/crates/rand/0.4.1 0.4.1 Apache-2.0 OR MIT
@@ -283,6 +290,7 @@
     https://crates.io/crates/scopeguard/1.1.0 1.1.0 Apache-2.0 OR MIT
     https://crates.io/crates/serde/1.0.143 1.0.143 Apache-2.0 OR MIT
     https://crates.io/crates/serde_derive/1.0.143 1.0.143 Apache-2.0 OR MIT
+    https://crates.io/crates/serde_json/1.0.0 1.0.0 Apache-2.0 OR MIT
     https://crates.io/crates/signal-hook-registry/1.1.1 1.1.1 Apache-2.0 OR MIT
     https://crates.io/crates/smallvec/1.6.1 1.6.1 Apache-2.0 OR MIT
     https://crates.io/crates/socket2/0.3.17 0.3.17 Apache-2.0 OR MIT
@@ -295,10 +303,12 @@
     https://crates.io/crates/thiserror-impl/1.0.32 1.0.32 Apache-2.0 OR MIT
     https://crates.io/crates/time/0.3.9 0.3.9 Apache-2.0 OR MIT
     https://crates.io/crates/tokio-io-timeout/1.0.1 1.0.1 Apache-2.0 OR MIT
+    https://crates.io/crates/toml/0.5.2 0.5.2 Apache-2.0 OR MIT
     https://crates.io/crates/unicode-segmentation/1.2.0 1.2.0 Apache-2.0 OR MIT
     https://crates.io/crates/unicode-width/0.1.4 0.1.4 Apache-2.0 OR MIT
     https://crates.io/crates/unicode-xid/0.2.0 0.2.0 Apache-2.0 OR MIT
     https://crates.io/crates/uuid/1.1.2 1.1.2 Apache-2.0 OR MIT
+    https://crates.io/crates/vcpkg/0.2.0 0.2.0 Apache-2.0 OR MIT
     https://crates.io/crates/vec_map/0.8.0 0.8.0 Apache-2.0 OR MIT
     https://crates.io/crates/version_check/0.9.0 0.9.0 Apache-2.0 OR MIT
     https://crates.io/crates/winapi/0.3.9 0.3.9 Apache-2.0 OR MIT
@@ -319,6 +329,8 @@
 
     https://crates.io/crates/fuchsia-zircon/0.3.1 0.3.1 BSD-3-Clause
     https://crates.io/crates/fuchsia-zircon-sys/0.3.1 0.3.1 BSD-3-Clause
+    https://crates.io/crates/num_enum/0.5.0 0.5.0 BSD-3-Clause
+    https://crates.io/crates/num_enum_derive/0.5.0 0.5.0 BSD-3-Clause
 
 ========================================================================
 MIT licenses
@@ -345,6 +357,8 @@
     https://crates.io/crates/mio/0.8.1 0.8.1 MIT
     https://crates.io/crates/nom/7.0.0 7.0.0 MIT
     https://crates.io/crates/proc-macro-error/0.2.0 0.2.0 MIT
+    https://crates.io/crates/rdkafka/0.32.2 0.32.2 MIT
+    https://crates.io/crates/rdkafka-sys/4.5.0+1.9.2 4.5.0+1.9.2 MIT
     https://crates.io/crates/redox_syscall/0.1.38 0.1.38 MIT
     https://crates.io/crates/redox_syscall/0.2.8 0.2.8 MIT
     https://crates.io/crates/slab/0.4.2 0.4.2 MIT
diff --git a/dist-material/licenses/LICENSE-rdkafka-sys.txt b/dist-material/licenses/LICENSE-rdkafka-sys.txt
new file mode 100644
index 0000000..eb7158a
--- /dev/null
+++ b/dist-material/licenses/LICENSE-rdkafka-sys.txt
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2016 Federico Giraud
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/dist-material/licenses/LICENSE-rdkafka.txt b/dist-material/licenses/LICENSE-rdkafka.txt
new file mode 100644
index 0000000..eb7158a
--- /dev/null
+++ b/dist-material/licenses/LICENSE-rdkafka.txt
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2016 Federico Giraud
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/docker-compose.e2e.yml b/docker-compose.e2e.yml
index 3462189..84c7d92 100644
--- a/docker-compose.e2e.yml
+++ b/docker-compose.e2e.yml
@@ -37,6 +37,8 @@
     depends_on:
       collector:
         condition: service_healthy
+      broker:
+        condition: service_healthy
     healthcheck:
       test: [ "CMD", "curl", "http://0.0.0.0:8082/healthCheck" ]
       interval: 5s
@@ -54,3 +56,15 @@
         condition: service_healthy
       consumer:
         condition: service_healthy
+      broker:
+        condition: service_healthy
+
+  broker:
+    image: landoop/fast-data-dev:3.3
+    container_name: broker
+    ports:
+      - "9092:9092"
+    healthcheck:
+      test: [ "CMD", "nc", "-zv", "0.0.0.0", "9092"]
+      interval: 5s
+      timeout: 5s
diff --git a/e2e/Cargo.toml b/e2e/Cargo.toml
index 2b156d0..d2e49e1 100644
--- a/e2e/Cargo.toml
+++ b/e2e/Cargo.toml
@@ -22,9 +22,16 @@
 edition = "2021"
 publish = false
 license = "Apache-2.0"
+default-run = "e2e"
+
+[[bin]]
+name = "e2e-kafka"
+path = "src/e2e_kafka.rs"
 
 [dependencies]
-skywalking = { path = ".." }
 hyper = { version = "0.14", features = ["full"] }
-tokio = { version = "1", features = ["full"] }
+prost = "0.11.0"
+rdkafka = "0.32.2"
+skywalking = { path = "..", features = ["kafka-reporter"] }
 structopt = "0.3"
+tokio = { version = "1", features = ["full"] }
diff --git a/e2e/docker/Dockerfile b/e2e/docker/Dockerfile
index 94ab5c3..13fb2d2 100644
--- a/e2e/docker/Dockerfile
+++ b/e2e/docker/Dockerfile
@@ -20,4 +20,5 @@
 WORKDIR /build
 COPY . /build/
 RUN cargo build --release --workspace
+ENV RUST_BACKTRACE=1
 ENTRYPOINT ["/build/target/release/e2e"]
diff --git a/e2e/src/e2e_kafka.rs b/e2e/src/e2e_kafka.rs
new file mode 100644
index 0000000..e785050
--- /dev/null
+++ b/e2e/src/e2e_kafka.rs
@@ -0,0 +1,293 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+#![allow(clippy::bool_assert_comparison)]
+
+use prost::Message;
+use rdkafka::{
+    consumer::{Consumer, StreamConsumer},
+    ClientConfig, Message as _,
+};
+use skywalking::proto::v3::{
+    log_data_body::Content, meter_data::Metric, JsonLog, KeyStringValuePair, LogData, LogTags,
+    MeterData, RefType, SegmentObject, SpanLayer, SpanType,
+};
+use std::time::Duration;
+use tokio::time::timeout;
+
+async fn segment() {
+    let consumer = create_consumer("skywalking-segments");
+
+    for _ in 0..3 {
+        let segment: SegmentObject = consumer_recv(&consumer).await;
+        check_segment(&segment);
+    }
+}
+
+fn check_segment(segment: &SegmentObject) {
+    dbg!(segment);
+
+    assert_eq!(segment.service_instance, "node_0");
+
+    if segment.service == "consumer" {
+        assert_eq!(segment.spans.len(), 1);
+
+        assert_eq!(segment.spans[0].span_id, 0);
+        assert_eq!(segment.spans[0].parent_span_id, -1);
+        assert_eq!(segment.spans[0].operation_name, "/pong");
+        assert_eq!(segment.spans[0].peer, "");
+        assert_eq!(segment.spans[0].span_type, SpanType::Entry as i32);
+        assert_eq!(segment.spans[0].span_layer, SpanLayer::Http as i32);
+        assert_eq!(segment.spans[0].component_id, 11000);
+        assert_eq!(segment.spans[0].is_error, false);
+        assert_eq!(segment.spans[0].tags.len(), 0);
+        assert_eq!(segment.spans[0].logs.len(), 0);
+        assert_eq!(segment.spans[0].refs.len(), 1);
+
+        assert_eq!(
+            segment.spans[0].refs[0].ref_type,
+            RefType::CrossProcess as i32
+        );
+        assert_eq!(segment.spans[0].refs[0].parent_span_id, 1);
+        assert_eq!(segment.spans[0].refs[0].parent_service, "producer");
+        assert_eq!(segment.spans[0].refs[0].parent_service_instance, "node_0");
+        assert_eq!(segment.spans[0].refs[0].parent_endpoint, "/pong");
+        assert_eq!(
+            segment.spans[0].refs[0].network_address_used_at_peer,
+            "consumer:8082"
+        );
+    } else if segment.service == "producer" {
+        if segment.spans.last().unwrap().operation_name == "/ping" {
+            assert_eq!(segment.spans.len(), 3);
+
+            assert_eq!(segment.spans[0].span_id, 1);
+            assert_eq!(segment.spans[0].parent_span_id, 0);
+            assert_eq!(segment.spans[0].operation_name, "/pong");
+            assert_eq!(segment.spans[0].peer, "consumer:8082");
+            assert_eq!(segment.spans[0].span_type, SpanType::Exit as i32);
+            assert_eq!(segment.spans[0].span_layer, SpanLayer::Unknown as i32);
+            assert_eq!(segment.spans[0].component_id, 11000);
+            assert_eq!(segment.spans[0].is_error, false);
+            assert_eq!(segment.spans[0].tags.len(), 0);
+            assert_eq!(segment.spans[0].logs.len(), 0);
+            assert_eq!(segment.spans[0].refs.len(), 0);
+
+            assert_eq!(segment.spans[1].span_id, 2);
+            assert_eq!(segment.spans[1].parent_span_id, 0);
+            assert_eq!(segment.spans[1].operation_name, "async-job");
+            assert_eq!(segment.spans[1].peer, "");
+            assert_eq!(segment.spans[1].span_type, SpanType::Local as i32);
+            assert_eq!(segment.spans[1].span_layer, SpanLayer::Unknown as i32);
+            assert_eq!(segment.spans[1].component_id, 11000);
+            assert_eq!(segment.spans[1].is_error, false);
+            assert_eq!(segment.spans[1].tags.len(), 0);
+            assert_eq!(segment.spans[1].logs.len(), 0);
+            assert_eq!(segment.spans[1].refs.len(), 0);
+
+            assert_eq!(segment.spans[2].span_id, 0);
+            assert_eq!(segment.spans[2].parent_span_id, -1);
+            assert_eq!(segment.spans[2].operation_name, "/ping");
+            assert_eq!(segment.spans[2].peer, "");
+            assert_eq!(segment.spans[2].span_type, SpanType::Entry as i32);
+            assert_eq!(segment.spans[2].span_layer, SpanLayer::Http as i32);
+            assert_eq!(segment.spans[2].component_id, 11000);
+            assert_eq!(segment.spans[2].is_error, false);
+            assert_eq!(segment.spans[2].tags.len(), 0);
+            assert_eq!(segment.spans[2].logs.len(), 0);
+            assert_eq!(segment.spans[2].refs.len(), 0);
+        } else if segment.spans.last().unwrap().operation_name == "async-callback" {
+            assert_eq!(segment.spans.len(), 1);
+
+            assert_eq!(segment.spans[0].span_id, 0);
+            assert_eq!(segment.spans[0].parent_span_id, -1);
+            assert_eq!(segment.spans[0].peer, "");
+            assert_eq!(segment.spans[0].span_type, SpanType::Entry as i32);
+            assert_eq!(segment.spans[0].span_layer, SpanLayer::Http as i32);
+            assert_eq!(segment.spans[0].component_id, 11000);
+            assert_eq!(segment.spans[0].is_error, false);
+            assert_eq!(segment.spans[0].tags.len(), 0);
+            assert_eq!(segment.spans[0].logs.len(), 0);
+            assert_eq!(segment.spans[0].refs.len(), 1);
+
+            assert_eq!(
+                segment.spans[0].refs[0].ref_type,
+                RefType::CrossThread as i32
+            );
+            assert_eq!(segment.spans[0].refs[0].parent_span_id, 2);
+            assert_eq!(segment.spans[0].refs[0].parent_service, "producer");
+            assert_eq!(segment.spans[0].refs[0].parent_service_instance, "node_0");
+            assert_eq!(segment.spans[0].refs[0].parent_endpoint, "async-job");
+            assert_eq!(segment.spans[0].refs[0].network_address_used_at_peer, "");
+        } else {
+            panic!(
+                "unknown operation_name {}",
+                segment.spans.last().unwrap().operation_name
+            );
+        }
+    } else {
+        panic!("unknown service {}", segment.service);
+    }
+}
+
+async fn meter() {
+    let consumer = create_consumer("skywalking-meters");
+
+    for _ in 0..3 {
+        let meter: MeterData = consumer_recv(&consumer).await;
+        check_meter(&meter);
+    }
+}
+
+fn check_meter(meter: &MeterData) {
+    dbg!(meter);
+
+    assert_eq!(meter.service, "consumer");
+    assert_eq!(meter.service_instance, "node_0");
+
+    match &meter.metric {
+        Some(Metric::SingleValue(value)) => {
+            assert_eq!(value.name, "instance_trace_count");
+
+            if value.labels[0].name == "region" && value.labels[0].value == "us-west" {
+                assert_eq!(value.labels[1].name, "az");
+                assert_eq!(value.labels[1].value, "az-1");
+                assert_eq!(value.value, 30.0);
+            } else if value.labels[0].name == "region" && value.labels[0].value == "us-east" {
+                assert_eq!(value.labels[1].name, "az");
+                assert_eq!(value.labels[1].value, "az-3");
+                assert_eq!(value.value, 20.0);
+            } else {
+                panic!("unknown label {:?}", &value.labels[0]);
+            }
+        }
+        Some(Metric::Histogram(value)) => {
+            assert_eq!(value.name, "instance_trace_count");
+            assert_eq!(value.labels[0].name, "region");
+            assert_eq!(value.labels[0].value, "us-north");
+            assert_eq!(value.labels[1].name, "az");
+            assert_eq!(value.labels[1].value, "az-1");
+            assert_eq!(value.values[0].bucket, 10.0);
+            assert_eq!(value.values[0].count, 1);
+            assert_eq!(value.values[1].bucket, 20.0);
+            assert_eq!(value.values[1].count, 2);
+            assert_eq!(value.values[2].bucket, 30.0);
+            assert_eq!(value.values[2].count, 0);
+        }
+        _ => {
+            panic!("unknown metric");
+        }
+    }
+}
+
+async fn log() {
+    let consumer = create_consumer("skywalking-logs");
+
+    for _ in 0..3 {
+        let log: LogData = consumer_recv(&consumer).await;
+        check_log(&log);
+    }
+}
+
+fn check_log(log: &LogData) {
+    dbg!(log);
+
+    if log.service == "producer" && log.service_instance == "node_0" {
+        assert_eq!(log.endpoint, "/ping");
+
+        match &log.body.as_ref().unwrap().content {
+            Some(Content::Json(json)) => {
+                assert_eq!(json.json, r#"{"message": "handle ping"}"#);
+                assert_eq!(log.trace_context, None);
+                assert_eq!(
+                    log.tags,
+                    Some(LogTags {
+                        data: vec![KeyStringValuePair {
+                            key: "level".to_string(),
+                            value: "DEBUG".to_string()
+                        }]
+                    })
+                );
+            }
+            Some(Content::Text(text)) => {
+                assert_eq!(text.text, "do http request");
+                assert_eq!(log.trace_context.as_ref().unwrap().span_id, 1);
+                assert_eq!(
+                    log.tags,
+                    Some(LogTags {
+                        data: vec![KeyStringValuePair {
+                            key: "level".to_string(),
+                            value: "INFO".to_string()
+                        }]
+                    })
+                );
+            }
+            body => {
+                panic!("unknown log body {:?}", body);
+            }
+        }
+    } else if log.service == "consumer" && log.service_instance == "node_0" {
+        assert_eq!(log.endpoint, "/pong");
+        assert_eq!(
+            log.body.as_ref().unwrap().content,
+            Some(Content::Json(JsonLog {
+                json: r#"{"message": "handle pong"}"#.to_string()
+            }))
+        );
+        assert_eq!(log.trace_context, None);
+        assert_eq!(
+            log.tags,
+            Some(LogTags {
+                data: vec![KeyStringValuePair {
+                    key: "level".to_string(),
+                    value: "DEBUG".to_string()
+                }]
+            })
+        );
+    } else {
+        panic!("unknown log {} {}", log.service, log.service_instance);
+    }
+}
+
+fn create_consumer(topic: &str) -> StreamConsumer {
+    let consumer: StreamConsumer = ClientConfig::new()
+        .set("bootstrap.servers", "broker:9092")
+        .set("broker.address.family", "v4")
+        .set("session.timeout.ms", "30000")
+        .set("enable.auto.commit", "true")
+        .set("auto.offset.reset", "earliest")
+        .set("enable.auto.offset.store", "true")
+        .set("group.id", topic)
+        .create()
+        .unwrap();
+    consumer.subscribe(&[topic]).unwrap();
+    consumer
+}
+
+async fn consumer_recv<T: Message + Default>(consumer: &StreamConsumer) -> T {
+    let message = timeout(Duration::from_secs(12), consumer.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    let value = message.payload_view::<[u8]>().unwrap().unwrap();
+    Message::decode(value).unwrap()
+}
+
+#[tokio::main(flavor = "multi_thread")]
+async fn main() {
+    segment().await;
+    meter().await;
+    log().await;
+}
diff --git a/e2e/src/main.rs b/e2e/src/main.rs
index 545c1c9..bf46cb3 100644
--- a/e2e/src/main.rs
+++ b/e2e/src/main.rs
@@ -29,7 +29,11 @@
         meter::{Counter, Gauge, Histogram},
         metricer::Metricer,
     },
-    reporter::grpc::GrpcReporter,
+    reporter::{
+        grpc::GrpcReporter,
+        kafka::{KafkaReportBuilder, KafkaReporter, RDKafkaClientConfig},
+        CollectItem, Report,
+    },
     trace::{
         propagation::{
             context::SKYWALKING_HTTP_CONTEXT_HEADER_KEY, decoder::decode_propagation,
@@ -40,6 +44,7 @@
 };
 use std::{convert::Infallible, error::Error, net::SocketAddr};
 use structopt::StructOpt;
+use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 
 static NOT_FOUND_MSG: &str = "not found";
 static SUCCESS_MSG: &str = "Success";
@@ -210,11 +215,52 @@
     mode: String,
 }
 
+#[derive(Clone)]
+struct CombineReporter {
+    grpc_reporter: GrpcReporter<UnboundedSender<CollectItem>, UnboundedReceiver<CollectItem>>,
+    kafka_reporter: KafkaReporter<UnboundedSender<CollectItem>>,
+}
+
+impl Report for CombineReporter {
+    fn report(&self, item: CollectItem) {
+        let typ = match &item {
+            CollectItem::Trace(_) => "trace",
+            CollectItem::Log(_) => "log",
+            CollectItem::Meter(_) => "meter",
+            _ => "unknown",
+        };
+        println!("report item type: {:?}", typ);
+        self.grpc_reporter.report(item.clone());
+        self.kafka_reporter.report(item);
+    }
+}
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     let opt = Opt::from_args();
-    let reporter = GrpcReporter::connect("http://collector:19876").await?;
-    let handle = reporter.reporting().await.spawn();
+
+    let reporter1 = GrpcReporter::connect("http://collector:19876").await?;
+    let handle1 = reporter1.reporting().await.spawn();
+
+    let mut client_config = RDKafkaClientConfig::new();
+    client_config
+        .set("bootstrap.servers", "broker:9092")
+        .set("message.timeout.ms", "6000");
+    let (reporter2, reporting) = KafkaReportBuilder::new(client_config)
+        .with_err_handle(|message, err| {
+            eprintln!(
+                "kafka reporter failed, message: {}, err: {:?}",
+                message, err
+            );
+        })
+        .build()
+        .await?;
+    let handle2 = reporting.spawn();
+
+    let reporter = CombineReporter {
+        grpc_reporter: reporter1,
+        kafka_reporter: reporter2,
+    };
 
     if opt.mode == "consumer" {
         tracer::set_global_tracer(Tracer::new("consumer", "node_0", reporter.clone()));
@@ -229,7 +275,8 @@
         unreachable!()
     }
 
-    handle.await?;
+    handle1.await?;
+    handle2.await?;
 
     Ok(())
 }
diff --git a/src/error/mod.rs b/src/error/mod.rs
index 2a26da6..b87a5b7 100644
--- a/src/error/mod.rs
+++ b/src/error/mod.rs
@@ -44,6 +44,11 @@
     #[error("tokio join failed: {0}")]
     TokioJoin(#[from] tokio::task::JoinError),
 
+    /// Kafka reporter error.
+    #[cfg(feature = "kafka-reporter")]
+    #[error("kafka reporter failed: {0}")]
+    KafkaReporter(crate::reporter::kafka::Error),
+
     /// Other uncovered errors.
     #[error(transparent)]
     Other(#[from] Box<dyn std::error::Error + Send + 'static>),
diff --git a/src/proto/v3/mod.rs b/src/proto/v3/mod.rs
index 0086949..d470966 100644
--- a/src/proto/v3/mod.rs
+++ b/src/proto/v3/mod.rs
@@ -18,6 +18,7 @@
 
 #![allow(missing_docs)]
 #![allow(rustdoc::invalid_html_tags)]
+#![allow(clippy::derive_partial_eq_without_eq)]
 
 use crate::common::system_time::{fetch_time, TimePeriod};
 
diff --git a/src/reporter/grpc.rs b/src/reporter/grpc.rs
index 5abb88f..31676e7 100644
--- a/src/reporter/grpc.rs
+++ b/src/reporter/grpc.rs
@@ -16,6 +16,7 @@
 
 //! Grpc implementation of [Report].
 
+use super::{CollectItemConsume, CollectItemProduce};
 #[cfg(feature = "management")]
 use crate::proto::v3::management_service_client::ManagementServiceClient;
 use crate::{
@@ -51,7 +52,6 @@
 };
 use tokio_stream::StreamExt;
 use tonic::{
-    async_trait,
     metadata::{Ascii, MetadataValue},
     service::{interceptor::InterceptedService, Interceptor},
     transport::{self, Channel, Endpoint},
@@ -71,66 +71,6 @@
     error!(?status, "{}", message);
 }
 
-/// Special purpose, used for user-defined production operations. Generally, it
-/// does not need to be handled.
-pub trait CollectItemProduce: Send + Sync + 'static {
-    /// Produce the collect item non-blocking.
-    fn produce(&self, item: CollectItem) -> Result<(), Box<dyn Error>>;
-}
-
-impl CollectItemProduce for () {
-    fn produce(&self, _item: CollectItem) -> Result<(), Box<dyn Error>> {
-        Ok(())
-    }
-}
-
-impl CollectItemProduce for mpsc::UnboundedSender<CollectItem> {
-    fn produce(&self, item: CollectItem) -> Result<(), Box<dyn Error>> {
-        Ok(self.send(item)?)
-    }
-}
-
-/// Special purpose, used for user-defined consume operations. Generally, it
-/// does not need to be handled.
-#[async_trait]
-pub trait CollectItemConsume: Send + Sync + 'static {
-    /// Consume the collect item blocking.
-    async fn consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>>;
-
-    /// Try to consume the collect item non-blocking.
-    async fn try_consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>>;
-}
-
-#[async_trait]
-impl CollectItemConsume for () {
-    async fn consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>> {
-        Ok(None)
-    }
-
-    async fn try_consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>> {
-        Ok(None)
-    }
-}
-
-#[async_trait]
-impl CollectItemConsume for mpsc::UnboundedReceiver<CollectItem> {
-    async fn consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>> {
-        Ok(self.recv().await)
-    }
-
-    async fn try_consume(&mut self) -> Result<Option<CollectItem>, Box<dyn Error + Send>> {
-        use mpsc::error::TryRecvError;
-
-        match self.try_recv() {
-            Ok(item) => Ok(Some(item)),
-            Err(e) => match e {
-                TryRecvError::Empty => Ok(None),
-                TryRecvError::Disconnected => Err(Box::new(e)),
-            },
-        }
-    }
-}
-
 #[derive(Default, Clone)]
 struct CustomInterceptor {
     authentication: Option<Arc<String>>,
diff --git a/src/reporter/kafka.rs b/src/reporter/kafka.rs
new file mode 100644
index 0000000..aff55ac
--- /dev/null
+++ b/src/reporter/kafka.rs
@@ -0,0 +1,348 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+//! Kafka implementation of [Report].
+
+use super::{CollectItemConsume, CollectItemProduce};
+use crate::reporter::{CollectItem, Report};
+pub use rdkafka::config::{ClientConfig as RDKafkaClientConfig, RDKafkaLogLevel};
+use rdkafka::producer::{FutureProducer, FutureRecord};
+use std::{
+    error,
+    future::{pending, Future},
+    pin::Pin,
+    sync::{
+        atomic::{AtomicBool, Ordering::Relaxed},
+        Arc,
+    },
+    time::Duration,
+};
+use tokio::{select, spawn, sync::mpsc, task::JoinHandle, try_join};
+use tracing::error;
+
+/// Kafka reporter error.
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    /// ksKafka error.
+    #[error(transparent)]
+    RdKafka(#[from] rdkafka::error::KafkaError),
+
+    /// kafka topic not found
+    #[error("topic not found: {topic}")]
+    TopicNotFound {
+        /// Name of kafka topic.
+        topic: String,
+    },
+}
+
+type DynErrHandler = dyn Fn(&str, &dyn error::Error) + Send + Sync + 'static;
+
+fn default_err_handle(message: &str, err: &dyn error::Error) {
+    error!(?err, "{}", message);
+}
+
+#[derive(Default)]
+struct State {
+    is_closing: AtomicBool,
+}
+
+impl State {
+    fn is_closing(&self) -> bool {
+        self.is_closing.load(Relaxed)
+    }
+}
+
+/// The Kafka reporter plugin support report traces, metrics, logs, instance
+/// properties to Kafka cluster.
+pub struct KafkaReportBuilder<P, C> {
+    state: Arc<State>,
+    producer: Arc<P>,
+    consumer: C,
+    client_config: RDKafkaClientConfig,
+    namespace: Option<String>,
+    err_handle: Arc<DynErrHandler>,
+}
+
+impl KafkaReportBuilder<mpsc::UnboundedSender<CollectItem>, mpsc::UnboundedReceiver<CollectItem>> {
+    /// Create builder, with rdkafka client configuration.
+    pub fn new(client_config: RDKafkaClientConfig) -> Self {
+        let (producer, consumer) = mpsc::unbounded_channel();
+        Self::new_with_pc(client_config, producer, consumer)
+    }
+}
+
+impl<P: CollectItemProduce, C: CollectItemConsume> KafkaReportBuilder<P, C> {
+    /// Special purpose, used for user-defined produce and consume operations,
+    /// usually you can use [KafkaReportBuilder::new].
+    pub fn new_with_pc(client_config: RDKafkaClientConfig, producer: P, consumer: C) -> Self {
+        Self {
+            state: Default::default(),
+            producer: Arc::new(producer),
+            consumer,
+            client_config,
+            namespace: None,
+            err_handle: Arc::new(default_err_handle),
+        }
+    }
+
+    /// Set error handle. By default, the error will be logged.
+    pub fn with_err_handle(
+        mut self,
+        handle: impl Fn(&str, &dyn error::Error) + Send + Sync + 'static,
+    ) -> Self {
+        self.err_handle = Arc::new(handle);
+        self
+    }
+
+    /// Use to isolate multi OAP server when using same Kafka cluster (final
+    /// topic name will append namespace before Kafka topics with - ).
+    pub fn with_namespace(mut self, namespace: impl Into<String>) -> Self {
+        self.namespace = Some(namespace.into());
+        self
+    }
+
+    /// Build the Reporter implemented [Report] in the foreground, and the
+    /// handle to push data to kafka in the background.
+    pub async fn build(self) -> Result<(KafkaReporter<P>, KafkaReporting<C>), Error> {
+        let kafka_producer = KafkaProducer::new(
+            self.client_config.create()?,
+            self.err_handle.clone(),
+            self.namespace,
+        )
+        .await?;
+        Ok((
+            KafkaReporter {
+                state: self.state.clone(),
+                producer: self.producer,
+                err_handle: self.err_handle,
+            },
+            KafkaReporting {
+                state: self.state,
+                consumer: self.consumer,
+                kafka_producer,
+                shutdown_signal: Box::pin(pending()),
+            },
+        ))
+    }
+}
+
+/// The kafka reporter implemented [Report].
+pub struct KafkaReporter<P> {
+    state: Arc<State>,
+    producer: Arc<P>,
+    err_handle: Arc<DynErrHandler>,
+}
+
+impl<P> Clone for KafkaReporter<P> {
+    #[inline]
+    fn clone(&self) -> Self {
+        Self {
+            state: self.state.clone(),
+            producer: self.producer.clone(),
+            err_handle: self.err_handle.clone(),
+        }
+    }
+}
+
+impl<P: CollectItemProduce> Report for KafkaReporter<P> {
+    fn report(&self, item: CollectItem) {
+        if !self.state.is_closing() {
+            if let Err(e) = self.producer.produce(item) {
+                (self.err_handle)("report collect item failed", &*e);
+            }
+        }
+    }
+}
+
+/// The handle to push data to kafka.
+pub struct KafkaReporting<C> {
+    state: Arc<State>,
+    consumer: C,
+    kafka_producer: KafkaProducer,
+    shutdown_signal: Pin<Box<dyn Future<Output = ()> + Send + Sync + 'static>>,
+}
+
+impl<C: CollectItemConsume> KafkaReporting<C> {
+    /// Quit when shutdown_signal received.
+    ///
+    /// Accept a `shutdown_signal` argument as a graceful shutdown signal.
+    pub fn with_graceful_shutdown(
+        mut self,
+        shutdown_signal: impl Future<Output = ()> + Send + Sync + 'static,
+    ) -> Self {
+        self.shutdown_signal = Box::pin(shutdown_signal);
+        self
+    }
+
+    /// Spawn the reporting in background.
+    pub fn spawn(self) -> ReportingJoinHandle {
+        let handle = spawn(async move {
+            let KafkaReporting {
+                state,
+                mut consumer,
+                mut kafka_producer,
+                shutdown_signal,
+            } = self;
+
+            let (shutdown_tx, mut shutdown_rx) = mpsc::unbounded_channel();
+
+            let work_fut = async move {
+                loop {
+                    select! {
+                        item = consumer.consume() => {
+                            match item {
+                                Ok(Some(item)) => {
+                                    kafka_producer.produce(item).await;
+                                }
+                                Ok(None) => break,
+                                Err(err) => return Err(crate::Error::Other(err)),
+                            }
+                        }
+                        _ =  shutdown_rx.recv() => break,
+                    }
+                }
+
+                state.is_closing.store(true, Relaxed);
+
+                // Flush.
+                loop {
+                    match consumer.try_consume().await {
+                        Ok(Some(item)) => {
+                            kafka_producer.produce(item).await;
+                        }
+                        Ok(None) => break,
+                        Err(err) => return Err(err.into()),
+                    }
+                }
+
+                Ok::<_, crate::Error>(())
+            };
+
+            let shutdown_fut = async move {
+                shutdown_signal.await;
+                shutdown_tx
+                    .send(())
+                    .map_err(|e| crate::Error::Other(Box::new(e)))?;
+                Ok(())
+            };
+
+            try_join!(work_fut, shutdown_fut)?;
+
+            Ok(())
+        });
+        ReportingJoinHandle { handle }
+    }
+}
+
+/// Handle of [KafkaReporting::spawn].
+pub struct ReportingJoinHandle {
+    handle: JoinHandle<crate::Result<()>>,
+}
+
+impl Future for ReportingJoinHandle {
+    type Output = crate::Result<()>;
+
+    fn poll(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Self::Output> {
+        Pin::new(&mut self.handle).poll(cx).map(|rs| rs?)
+    }
+}
+
+struct TopicNames {
+    segment: String,
+    meter: String,
+    log: String,
+    #[cfg(feature = "management")]
+    management: String,
+}
+
+impl TopicNames {
+    const TOPIC_LOG: &str = "skywalking-logs";
+    #[cfg(feature = "management")]
+    const TOPIC_MANAGEMENT: &str = "skywalking-managements";
+    const TOPIC_METER: &str = "skywalking-meters";
+    const TOPIC_SEGMENT: &str = "skywalking-segments";
+
+    fn new(namespace: Option<&str>) -> Self {
+        Self {
+            segment: Self::real_topic_name(namespace, Self::TOPIC_SEGMENT),
+            meter: Self::real_topic_name(namespace, Self::TOPIC_METER),
+            log: Self::real_topic_name(namespace, Self::TOPIC_LOG),
+            #[cfg(feature = "management")]
+            management: Self::real_topic_name(namespace, Self::TOPIC_MANAGEMENT),
+        }
+    }
+
+    fn real_topic_name(namespace: Option<&str>, topic_name: &str) -> String {
+        namespace
+            .map(|namespace| format!("{}-{}", namespace, topic_name))
+            .unwrap_or_else(|| topic_name.to_string())
+    }
+}
+
+struct KafkaProducer {
+    topic_names: TopicNames,
+    client: FutureProducer,
+    err_handle: Arc<DynErrHandler>,
+}
+
+impl KafkaProducer {
+    async fn new(
+        client: FutureProducer,
+        err_handle: Arc<DynErrHandler>,
+        namespace: Option<String>,
+    ) -> Result<Self, Error> {
+        let topic_names = TopicNames::new(namespace.as_deref());
+        Ok(Self {
+            client,
+            err_handle,
+            topic_names,
+        })
+    }
+
+    async fn produce(&mut self, item: CollectItem) {
+        let (topic_name, key) = match &item {
+            CollectItem::Trace(item) => (
+                &self.topic_names.segment,
+                item.trace_segment_id.as_bytes().to_vec(),
+            ),
+            CollectItem::Log(item) => (&self.topic_names.log, item.service.as_bytes().to_vec()),
+            CollectItem::Meter(item) => (
+                &self.topic_names.meter,
+                item.service_instance.as_bytes().to_vec(),
+            ),
+            #[cfg(feature = "management")]
+            CollectItem::Instance(item) => (
+                &self.topic_names.management,
+                format!("register-{}", &item.service_instance).into_bytes(),
+            ),
+            #[cfg(feature = "management")]
+            CollectItem::Ping(item) => (
+                &self.topic_names.log,
+                item.service_instance.as_bytes().to_vec(),
+            ),
+        };
+
+        let payload = item.encode_to_vec();
+        let record = FutureRecord::to(topic_name).payload(&payload).key(&key);
+
+        if let Err((err, _)) = self.client.send(record, Duration::from_secs(0)).await {
+            (self.err_handle)("Collect data to kafka failed", &err);
+        }
+    }
+}
diff --git a/src/reporter/mod.rs b/src/reporter/mod.rs
index 0b46662..783fcfe 100644
--- a/src/reporter/mod.rs
+++ b/src/reporter/mod.rs
@@ -17,17 +17,21 @@
 //! Reporter contains common `Report` trait and the implementations.
 
 pub mod grpc;
+#[cfg(feature = "kafka-reporter")]
+#[cfg_attr(docsrs, doc(cfg(feature = "kafka-reporter")))]
+pub mod kafka;
 pub mod print;
 
 #[cfg(feature = "management")]
 use crate::proto::v3::{InstancePingPkg, InstanceProperties};
 use crate::proto::v3::{LogData, MeterData, SegmentObject};
 use serde::{Deserialize, Serialize};
-use std::{ops::Deref, sync::Arc};
-use tokio::sync::OnceCell;
+use std::{error::Error, ops::Deref, sync::Arc};
+use tokio::sync::{mpsc, OnceCell};
+use tonic::async_trait;
 
 /// Collect item of protobuf object.
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 #[non_exhaustive]
 pub enum CollectItem {
     /// Tracing object.
@@ -46,6 +50,23 @@
     Ping(Box<InstancePingPkg>),
 }
 
+impl CollectItem {
+    #[cfg(feature = "kafka-reporter")]
+    pub(crate) fn encode_to_vec(self) -> Vec<u8> {
+        use prost::Message;
+
+        match self {
+            CollectItem::Trace(item) => item.encode_to_vec(),
+            CollectItem::Log(item) => item.encode_to_vec(),
+            CollectItem::Meter(item) => item.encode_to_vec(),
+            #[cfg(feature = "management")]
+            CollectItem::Instance(item) => item.encode_to_vec(),
+            #[cfg(feature = "management")]
+            CollectItem::Ping(item) => item.encode_to_vec(),
+        }
+    }
+}
+
 pub(crate) type DynReport = dyn Report + Send + Sync + 'static;
 
 /// Report provide non-blocking report method for trace, log and metric object.
@@ -76,3 +97,91 @@
         Report::report(self.get().expect("OnceCell is empty"), item)
     }
 }
+
+/// Special purpose, used for user-defined production operations. Generally, it
+/// does not need to be handled.
+pub trait CollectItemProduce: Send + Sync + 'static {
+    /// Produce the collect item non-blocking.
+    fn produce(&self, item: CollectItem) -> Result<(), Box<dyn Error>>;
+}
+
+impl CollectItemProduce for () {
+    fn produce(&self, _item: CollectItem) -> Result<(), Box<dyn Error>> {
+        Ok(())
+    }
+}
+
+impl CollectItemProduce for mpsc::Sender<CollectItem> {
+    fn produce(&self, item: CollectItem) -> Result<(), Box<dyn Error>> {
+        Ok(self.blocking_send(item)?)
+    }
+}
+
+impl CollectItemProduce for mpsc::UnboundedSender<CollectItem> {
+    fn produce(&self, item: CollectItem) -> Result<(), Box<dyn Error>> {
+        Ok(self.send(item)?)
+    }
+}
+
+/// Alias of method result of [CollectItemConsume].
+pub type ConsumeResult = Result<Option<CollectItem>, Box<dyn Error + Send>>;
+
+/// Special purpose, used for user-defined consume operations. Generally, it
+/// does not need to be handled.
+#[async_trait]
+pub trait CollectItemConsume: Send + Sync + 'static {
+    /// Consume the collect item blocking.
+    async fn consume(&mut self) -> ConsumeResult;
+
+    /// Try to consume the collect item non-blocking.
+    async fn try_consume(&mut self) -> ConsumeResult;
+}
+
+#[async_trait]
+impl CollectItemConsume for () {
+    async fn consume(&mut self) -> ConsumeResult {
+        Ok(None)
+    }
+
+    async fn try_consume(&mut self) -> ConsumeResult {
+        Ok(None)
+    }
+}
+
+#[async_trait]
+impl CollectItemConsume for mpsc::Receiver<CollectItem> {
+    async fn consume(&mut self) -> ConsumeResult {
+        Ok(self.recv().await)
+    }
+
+    async fn try_consume(&mut self) -> ConsumeResult {
+        use mpsc::error::TryRecvError;
+
+        match self.try_recv() {
+            Ok(item) => Ok(Some(item)),
+            Err(e) => match e {
+                TryRecvError::Empty => Ok(None),
+                TryRecvError::Disconnected => Err(Box::new(e)),
+            },
+        }
+    }
+}
+
+#[async_trait]
+impl CollectItemConsume for mpsc::UnboundedReceiver<CollectItem> {
+    async fn consume(&mut self) -> ConsumeResult {
+        Ok(self.recv().await)
+    }
+
+    async fn try_consume(&mut self) -> ConsumeResult {
+        use mpsc::error::TryRecvError;
+
+        match self.try_recv() {
+            Ok(item) => Ok(Some(item)),
+            Err(e) => match e {
+                TryRecvError::Empty => Ok(None),
+                TryRecvError::Disconnected => Err(Box::new(e)),
+            },
+        }
+    }
+}