Merge branch 'master' into DISCLAIMER

.github/ISSUE_TEMPLATE

 Please answer these questions before submitting your issue.
 
-### What version of sky-walking are you using?
+- Why do you submit this issue?
+- [ ] Question or discussion
+- [ ] Bug
+- [ ] Requirement
+- [ ] Feature or performance improvement
 
+___
+### Question
+- What do you want to know?
 
-### What version of your OS?
+___
+### Bug
+- Which version of SkyWalking, OS and JRE?
 
+- Which company or project?
 
-### What version of your JRE?
-
-
-### What company or project?
-
-
-### What did you do?
+- What happen?
 If possible, provide a way for reproducing the error. e.g. demo application, component version.
+
+___
+### Requirement or improvement
+- Please describe about your requirements or improvement suggestions.
\ No newline at end of file

.github/PULL_REQUEST_TEMPLATE

+Please answer these questions before submitting pull request
+
+- Why submit this pull request?
+- [ ] Bug fix
+- [ ] New feature provided
+- [ ] Improve performance
+
+- Related issues
+
+___
+### Bug fix
+- Bug description.
+
+- How to fix?
+
+___
+### New feature or improvement
+- Describe the details and related test reports.

README.md

@@ -7,8 +7,10 @@ Apache SkyWalking | [中文](README_ZH.md)
 microservices, cloud native and container-based (Docker, K8s, Mesos) architectures.
 Underlying technology is a distributed tracing system.
 
+[![GitHub stars](https://img.shields.io/github/stars/apache/incubator-skywalking.svg?style=for-the-badge&label=Stars&logo=github)](https://github.com/apache/incubator-skywalking)
+[![Twitter Follow](https://img.shields.io/twitter/follow/asfskywalking.svg?style=for-the-badge&label=Follow&logo=twitter)](https://twitter.com/AsfSkyWalking)
+
 [![Build Status](https://travis-ci.org/apache/incubator-skywalking.svg?branch=master)](https://travis-ci.org/apache/incubator-skywalking)
-[![Twitter URL](https://img.shields.io/twitter/url/http/shields.io.svg?style=social)](https://twitter.com/AsfSkyWalking)
 [![Join the chat at https://gitter.im/sky-walking/Lobby](https://badges.gitter.im/openskywalking/Lobby.svg)](https://gitter.im/openskywalking/Lobby)
 [![OpenTracing-1.x Badge](https://img.shields.io/badge/OpenTracing--1.x-enabled-blue.svg)](http://opentracing.io)
 

README_ZH.md

@@ -6,8 +6,10 @@ Apache SkyWalking | [English](README.md)
 **SkyWalking**: 针对分布式系统的APM（应用性能监控）系统，特别针对微服务、cloud native和容器化(Docker, K8s, Mesos)架构，
 其核心是个分布式追踪系统。
 
+[![GitHub stars](https://img.shields.io/github/stars/apache/incubator-skywalking.svg?style=for-the-badge&label=Stars&logo=github)](https://github.com/apache/incubator-skywalking)
+[![Twitter Follow](https://img.shields.io/twitter/follow/asfskywalking.svg?style=for-the-badge&label=Follow&logo=twitter)](https://twitter.com/AsfSkyWalking)
+
 [![Build Status](https://travis-ci.org/apache/incubator-skywalking.svg?branch=master)](https://travis-ci.org/apache/incubator-skywalking)
-[![Twitter URL](https://img.shields.io/twitter/url/http/shields.io.svg?style=social)](https://twitter.com/AsfSkyWalking)
 [![Join the chat at https://gitter.im/openskywalking/Lobby](https://badges.gitter.im/openskywalking/Lobby.svg)](https://gitter.im/openskywalking/Lobby)
 [![OpenTracing-1.x Badge](https://img.shields.io/badge/OpenTracing--1.x-enabled-blue.svg)](http://opentracing.io)
 

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostSpanListener.java

@@ -33,6 +33,7 @@ import org.apache.skywalking.apm.collector.cache.service.ServiceNameCacheService
 import org.apache.skywalking.apm.collector.core.graph.Graph;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.core.util.BooleanUtils;
 import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
 import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost;
 import org.slf4j.Logger;
@@ -98,7 +99,7 @@ public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListe
         Graph<SegmentCost> graph = GraphManager.INSTANCE.findGraph(MetricGraphIdDefine.SEGMENT_COST_GRAPH_ID, SegmentCost.class);
         logger.debug("segment cost listener build");
         for (SegmentCost segmentCost : segmentCosts) {
-            segmentCost.setIsError(isError);
+            segmentCost.setIsError(BooleanUtils.booleanToValue(isError));
             segmentCost.setTimeBucket(timeBucket);
             graph.start(segmentCost);
         }

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/IdAutoIncrement.java

@@ -25,18 +25,20 @@ public enum IdAutoIncrement {
     INSTANCE;
 
     public int increment(int min, int max) {
-        int instanceId;
+        int id;
         if (min == max) {
-            instanceId = -1;
+            if (min == 0) {
+                id = -1;
+            } else {
+                id = 1;
+            }
         } else if (min + max == 0) {
-            instanceId = max + 1;
+            id = max + 1;
         } else if (min + max > 0) {
-            instanceId = min - 1;
-        } else if (max < 0) {
-            instanceId = 1;
+            id = min - 1;
         } else {
-            instanceId = max + 1;
+            id = max + 1;
         }
-        return instanceId;
+        return id;
     }
 }

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/test/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/IdAutoIncrementTestCase.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.analysis.register.provider.register;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * @author peng-yongsheng
+ */
+public class IdAutoIncrementTestCase {
+
+    @Test
+    public void testIncrement() {
+        int id = IdAutoIncrement.INSTANCE.increment(0, 0);
+        Assert.assertEquals(-1, id);
+
+        id = IdAutoIncrement.INSTANCE.increment(-1, -1);
+        Assert.assertEquals(1, id);
+
+        id = IdAutoIncrement.INSTANCE.increment(-1, 1);
+        Assert.assertEquals(2, id);
+
+        id = IdAutoIncrement.INSTANCE.increment(-1, 2);
+        Assert.assertEquals(-2, id);
+
+        id = IdAutoIncrement.INSTANCE.increment(-2, 2);
+        Assert.assertEquals(3, id);
+    }
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/segment/SegmentCost.java

@@ -22,7 +22,6 @@ import org.apache.skywalking.apm.collector.core.data.Column;
 import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.data.operator.CoverOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
-import org.apache.skywalking.apm.collector.core.util.BooleanUtils;
 
 /**
  * @author peng-yongsheng
@@ -127,11 +126,11 @@ public class SegmentCost extends StreamData {
         setDataInteger(0, applicationId);
     }
 
-    public Boolean getIsError() {
-        return BooleanUtils.valueToBoolean(getDataInteger(1));
+    public Integer getIsError() {
+        return getDataInteger(1);
     }
 
-    public void setIsError(Boolean isError) {
-        setDataInteger(0, BooleanUtils.booleanToValue(isError));
+    public void setIsError(Integer isError) {
+        setDataInteger(1, isError);
     }
 }

apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java

@@ -76,6 +76,7 @@ public class ComponentsDefine {
   
     public static final OfficialComponent ROCKET_MQ = new OfficialComponent(25, "RocketMQ");
 
+    public static final OfficialComponent HTTP_ASYNC_CLIENT = new OfficialComponent(26, "httpasyncclient");
 
     private static ComponentsDefine INSTANCE = new ComponentsDefine();
 
@@ -86,7 +87,7 @@ public class ComponentsDefine {
     }
 
     public ComponentsDefine() {
-        components = new String[26];
+        components = new String[27];
         addComponent(TOMCAT);
         addComponent(HTTPCLIENT);
         addComponent(DUBBO);
@@ -112,6 +113,7 @@ public class ComponentsDefine {
         addComponent(GRPC);
         addComponent(ELASTIC_JOB);
         addComponent(ROCKET_MQ);
+        addComponent(HTTP_ASYNC_CLIENT);
     }
 
     private void addComponent(OfficialComponent component) {

apm-protocol/apm-ui-protocol/src/main/resources/ui-graphql/alarm.graphqls

+type Alarm {
+    items: [AlarmItem!]!
+    count: Int!
+}
+
 type AlarmItem {
+    # Typical include: Application Code + cause type. This is a short description.
+    title: String!
+    # Include all related info to trigger this alarm.
+    # such as: threshold, trigger value, relation(greater or lower), last time
     content: String!
     startTime: String!
     alertType: AlarmType!
+    causeType: CauseType!
 }
 
 enum AlarmType {
@@ -10,6 +20,11 @@ enum AlarmType {
     SERVICE
 }
 
+enum CauseType {
+    LOW_SUCCESS_RATE,
+    SLOW_RESPONSE
+}
+
 extend type Query {
-    loadAlertList(keyword: String, alertType: AlarmType, duration:Duration!):[AlarmItem]
+    loadAlertList(keyword: String, alertType: AlarmType, duration:Duration!, paging: Pagination!): Alarm
 }
\ No newline at end of file

apm-protocol/apm-ui-protocol/src/main/resources/ui-graphql/application-layer.graphqls

@@ -8,20 +8,22 @@ type ApplicationNode implements Node {
     # 2 Digits after floating point.
     sla: Float!
     # The number of incoming calls
-    calls: Long!
+    callsPerSec: Long!
+    # Unit: millisecond
+    responseTimePerSec: Int!
     # ref: http://www.apdex.org/
     # Max value is 1
     # 2 Digits after floating point.
     apdex: Float!
+    # Whether the application alerts?
+    # Default value is false.
+    isAlarm: Boolean!
     # The number of servers in the application code
     numOfServer: Int!
     # The number of servers alerting
     numOfServerAlarm: Int!
     # The number of services alerting
     numOfServiceAlarm: Int!
-    # Incoming request node, means User or outside system access the cluster from this.
-    # Recommend the UI generate a User node for each incoming node
-    isIncomingNode: Boolean
 }
 
 # The conjectural node generated by exit span
@@ -33,8 +35,8 @@ type ConjecturalNode implements Node {
 
 
 extend type Query {
-  getAllApplication(duration: Duration!): [ApplicationNode]
+  getAllApplication(duration: Duration!): [ApplicationNode!]!
   getApplicationTopology(applicationId: ID!, duration: Duration!): Topology
-  getSlowService(applicationId: ID!, duration: Duration!): [ServiceInfo!]
-  getServerThroughput(applicationId: ID!, duration: Duration!): [AppServerInfo!]
+  getSlowService(applicationId: ID!, duration: Duration!, top: Int!): [ServiceInfo!]!
+  getServerThroughput(applicationId: ID!, duration: Duration!, top: Int!): [AppServerInfo!]!
 }

apm-protocol/apm-ui-protocol/src/main/resources/ui-graphql/common.graphqls

 schema {
     query: Query
+    mutation: Mutation
 }
 
 #Root node
@@ -7,6 +8,10 @@ type Query {
     version: String
 }
 
+type Mutation {
+    version: String
+}
+
 # The Duration defines the start and end time for each query operation.
 # Fields: `start` and `end`
 #   represents the time span. And each of them matches the step.
@@ -39,6 +44,14 @@ enum Step {
     SECOND
 }
 
+input Pagination {
+    # pageNum starts in 1, the default is 1.
+    pageNum: Int
+    pageSize: Int!
+    # default false
+    needTotal: Boolean
+}
+
 ######################################
 # Common Metrics and Trends
 ######################################
@@ -47,17 +60,17 @@ type ResponseTimeTrend {
 }
 
 type ThroughputTrend {
-    trendList: [Int!]
+    trendList: [Int!]!
 }
 
 type SLATrend {
-    trendList: [Int!]
+    trendList: [Int!]!
 }
 
 # The overview topology of the whole application cluster or services,
 type Topology {
-    nodes: [Node]!
-    calls: [Call]
+    nodes: [Node!]!
+    calls: [Call!]!
 }
 
 # The base Node of all node types in topology
@@ -74,6 +87,15 @@ interface Node {
     type: String
 }
 
+# Incoming request node, means User or outside system access the cluster from this.
+type VisualUserNode implements Node {
+    id: ID!
+    # Constant, value = "User"
+    name: String!
+    # Constant, value = "USER"
+    type: String
+}
+
 # The Call represents a directed distributed call,
 # from the `source` to the `target`.
 type Call {

apm-protocol/apm-ui-protocol/src/main/resources/ui-graphql/config.graphqls

+input TTLConfigItem {
+    unit: Step!
+    value: Int!
+}
+
+type ExistedTTLConfigs{
+    ttl: [TTL!]!
+}
+
+type TTL {
+     unit: Step!
+     value: Int!
+}
+
+input AlarmThreshold {
+    type: AlarmType!
+    threshold: Int!
+    causeType: CauseType!
+}
+
+type ExistedAlarmThresholds {
+    items: [ExistedAlarmThresholdItem!]!
+}
+
+type ExistedAlarmThresholdItem {
+    threshold: Int!
+    causeType: CauseType!
+}
+
+extend type Mutation {
+    setDataTTLConfigs(ttl: [TTLConfigItem!]!): Boolean!
+    setAlarmThreshold(thresholds: [AlarmThreshold!]!): Boolean!
+}
+
+extend type Query {
+    queryAllDataTTLConfigs: ExistedTTLConfigs!
+    queryAlarmThresholds(alarmType: AlarmType): ExistedAlarmThresholds!
+}
\ No newline at end of file

apm-protocol/apm-ui-protocol/src/main/resources/ui-graphql/overview-layer.graphqls

@@ -15,7 +15,7 @@ type AlarmTrend {
 # Query all conjectural applications based on the given duration
 # All applications here are not installed agent.
 type ConjecturalAppBrief {
-    apps: [ConjecturalApp!]
+    apps: [ConjecturalApp!]!
 }
 
 # The basic info of the conjectural application,
@@ -32,6 +32,6 @@ extend type Query {
     getClusterBrief(duration: Duration!): ClusterBrief
     getAlarmTrend(duration: Duration!): AlarmTrend
     getConjecturalApps(duration: Duration!): ConjecturalAppBrief
-    getTopNSlowService(duration: Duration!, topN: Int!): [ServiceInfo!]
-    getTopNServerThroughput(duration: Duration!, topN: Int!): [AppServerInfo!]
+    getTopNSlowService(duration: Duration!, topN: Int!): [ServiceInfo!]!
+    getTopNServerThroughput(duration: Duration!, topN: Int!): [AppServerInfo!]!
 }

apm-protocol/apm-ui-protocol/src/main/resources/ui-graphql/server-layer.graphqls

@@ -10,30 +10,31 @@ type AppServerInfo {
     os: String
     host: String
     pid: Int
-    IPv4: String
-    IPv6: String
+    ipv4: String
+    ipv6: String
 }
 
 type CPUTrend {
-    cost: [Int!]
+    cost: [Int!]!
 }
 
 # The gc trend represents the numbers of Garbage Collector execution
 type GCTrend {
-    youngGC: [Int!]
-    oldGC: [Int!]
+    youngGC: [Int!]!
+    oldGC: [Int!]!
 }
 
 # The memory used and max limit in heap and noheap space.
 type MemoryTrend {
-    heap: [Int!]
-    maxHeap: [Int!]
-    noheap: [Int!]
-    maxNoheap: [Int!]
+    heap: [Int!]!
+    maxHeap: [Int!]!
+    noheap: [Int!]!
+    maxNoheap: [Int!]!
 }
 
 extend type Query {
-    searchServer(keyword: String!, duration: Duration!): [AppServerInfo]
+    searchServer(keyword: String!, duration: Duration!): [AppServerInfo!]!
+    getAllServer(applicationId: ID!, duration: Duration!): [AppServerInfo!]!
     getServerResponseTimeTrend(serverId: ID!, duration: Duration!): ResponseTimeTrend
     getServerTPSTrend(serverId: ID!, duration: Duration!): ThroughputTrend
     getCPUTrend(serverId: ID!, duration: Duration!): CPUTrend

apm-protocol/apm-ui-protocol/src/main/resources/ui-graphql/service-layer.graphqls

@@ -27,7 +27,7 @@ type TraceItem {
 }
 
 extend type Query {
-    searchService(keyword: String!, duration: Duration!): [ServiceNode]
+    searchService(keyword: String!, duration: Duration!, topN: Int!): [ServiceNode!]!
     getServiceResponseTimeTrend(serviceId: ID!, duration: Duration!): ResponseTimeTrend
     getServiceTPSTrend(serviceId: ID!, duration: Duration!): ThroughputTrend
     getServiceSLATrend(serviceId: ID!, duration: Duration!): SLATrend

apm-protocol/apm-ui-protocol/src/main/resources/ui-graphql/trace.graphqls

 # The list of traces
 type TraceBrief {
-  traces: [BasicTrace!]
+  traces: [BasicTrace!]!
+  total: Int!
 }
 
 # Trace basic info
@@ -23,8 +24,7 @@ input TraceQueryCondition {
   minTraceDuration: Int
   # The max time of trace
   maxTraceDuration: Int
-  topN: Boolean
-  needTotal: Int
+  paging: Pagination!
 }
 
 enum QueryOrder {
@@ -34,21 +34,15 @@ enum QueryOrder {
 
 # The trace represents a distributed trace, includes all segments and spans.
 type Trace {
-  traceId: ID!
-  segments: [Segment!]
-}
-
-type Segment {
-  segmentId: ID!
-  appName: String!
-  isSizeLimited: Boolean!
   spans: [Span!]!
 }
 
 type Span {
-  refs: [Ref!]
+  traceId: ID!
+  segmentId: ID!
   spanId: Int!
   parentSpanId: Int!
+  refs: [Ref!]!
   startTime: Long!
   endTime: Long!
   operationName: String
@@ -60,17 +54,25 @@ type Span {
   isError: Boolean
   # There are 5 layers: Unknown, Database, RPCFramework, Http, MQ and Cache
   layer: String
-  tags: [KeyValue!]
-  logs: [LogEntity!]
+  tags: [KeyValue!]!
+  logs: [LogEntity!]!
 }
 
 # Ref represents the link between the segment and its parents.
+# The parent(ref) may not exists, which means batch process.
+# The UI should display a list, representing the other trace IDs.
 type Ref {
+  traceId: ID!
   parentSegmentId: ID!
   parentSpanId: Int!
   # Ref type represents why did the ref happen.
   # Include: 1) CrossProcess 2) CrossThread
-  type: String!
+  type: RefType!
+}
+
+enum RefType {
+  CROSS_PROCESS,
+  CROSS_THREAD
 }
 
 type KeyValue {
@@ -86,4 +88,4 @@ type LogEntity {
 extend type Query {
    queryBasicTraces(condition: TraceQueryCondition): TraceBrief
    queryTrace(traceId: ID!): Trace
-}
\ No newline at end of file
+}

apm-protocol/apm-ui-protocol/src/test/java/org/apache/skywalking/apm/ui/protocol/GraphQLScriptTest.java

@@ -43,6 +43,7 @@ public class GraphQLScriptTest {
         typeRegistry.merge(schemaParser.parse(loadSchema("server-layer.graphqls")));
         typeRegistry.merge(schemaParser.parse(loadSchema("service-layer.graphqls")));
         typeRegistry.merge(schemaParser.parse(loadSchema("alarm.graphqls")));
+        typeRegistry.merge(schemaParser.parse(loadSchema("config.graphqls")));
         RuntimeWiring wiring = buildRuntimeWiring();
         assertTrue(schemaGenerator.makeExecutableSchema(typeRegistry, wiring).getAllTypesAsList().size() > 0);
     }

apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/context/tag/Tags.java

@@ -59,6 +59,16 @@ public final class Tags {
      */
     public static final StringTag DB_BIND_VARIABLES = new StringTag("db.bind_vars");
 
+    /**
+     * MQ_BROKER records the broker address of message-middleware
+     */
+    public static final StringTag MQ_BROKER = new StringTag("mq.broker");
+
+    /**
+     * MQ_TOPIC records the topic name of message-middleware
+     */
+    public static final StringTag MQ_TOPIC = new StringTag("mq.topic");
+
     public static final class HTTP {
         public static final StringTag METHOD = new StringTag("http.method");
     }

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/pom.xml

+<?xml version="1.0"?>
+<!--
+  ~ 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.
+  ~
+  -->
+
+<project
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+    xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.skywalking</groupId>
+        <artifactId>apm-sdk-plugin</artifactId>
+        <version>5.0.0-alpha</version>
+    </parent>
+
+    <artifactId>apm-httpasyncclient-4.x-plugin</artifactId>
+
+    <name>httpasyncclient-4.x-plugin</name>
+    <url>http://maven.apache.org</url>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpasyncclient</artifactId>
+            <version>4.1.1</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpasyncclient-cache</artifactId>
+            <version>4.1.1</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <!-- 源码插件 -->
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-source-plugin</artifactId>
+                <!-- 发布时自动将源码同时发布的配置 -->
+                <executions>
+                    <execution>
+                        <id>attach-sources</id>
+                        <goals>
+                            <goal>jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/ConnectIterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+
+/**
+ * Pass ref accross thread by SessionRequest.
+ *
+ * @author liyuntao
+ */
+
+public class ConnectIterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        ((EnhancedInstance)ret).setSkyWalkingDynamicField(ContextManager.capture());
+
+        InetSocketAddress remoteAddress = (InetSocketAddress)allArguments[0];
+        String peer = remoteAddress.toString().substring(1);
+
+        Object[] cacheValue = new Object[3];
+        cacheValue[0] = ContextManager.capture();
+        cacheValue[1] = peer;
+        objInst.setSkyWalkingDynamicField(cacheValue);
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/DefaultConnectingIOReactorIterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * Create local span :httpasyncclient/SocketChannel, to showcase the ability to connect to the remote host.
+ *
+ * @author liyuntao
+ */
+
+public class DefaultConnectingIOReactorIterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        Object[] cacheValue = (Object[])objInst.getSkyWalkingDynamicField();
+        final ContextCarrier contextCarrier = new ContextCarrier();
+        AbstractSpan span = ContextManager.createExitSpan("httpasyncclient/" + method.getName(), contextCarrier, cacheValue[1].toString());
+        ContextManager.continued((ContextSnapshot)cacheValue[0]);
+        span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        ContextManager.stopSpan();
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/HttpAsyncResponseConsumerInterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.http.nio.protocol.HttpAsyncRequestProducer;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * End a local span for  {@link org.apache.http.impl.nio.client.CloseableHttpAsyncClient#execute} called by
+ * application.
+ *
+ * @author liyuntao
+ */
+
+public class HttpAsyncResponseConsumerInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+
+        HttpAsyncRequestProducer producer = (HttpAsyncRequestProducer)allArguments[0];
+        String uri = producer.generateRequest().getRequestLine().getUri();
+        String requestMethod = producer.generateRequest().getRequestLine().getMethod();
+        AbstractSpan span = ContextManager.createLocalSpan("httpasyncclient/" + method.getName());
+        Tags.HTTP.METHOD.set(span, requestMethod);
+        span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+        Tags.URL.set(span, uri);
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        ContextManager.stopSpan();
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/HttpHostInterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * End a local span for  {@link org.apache.http.impl.nio.client.CloseableHttpAsyncClient#execute} called by
+ * application.
+ *
+ * @author liyuntao
+ */
+
+public class HttpHostInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+
+        HttpHost producer = (HttpHost)allArguments[0];
+        String uri = producer.toURI();
+        AbstractSpan span = ContextManager.createLocalSpan("httpasyncclient/" + method.getName());
+        span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+        Tags.HTTP.METHOD.set(span, ((HttpRequest)allArguments[1]).getRequestLine().getMethod());
+        Tags.URL.set(span, uri);
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        ContextManager.stopSpan();
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/ProcessResponseInterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * 1.End exit span.
+ * 2.Create a local span of  callback.
+ * 3.End local span:AsyncThread/execute.
+ *
+ * @author liyuntao
+ */
+
+public class ProcessResponseInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        String uri = activeSpan.getOperationName();
+        //stop exitSpan
+        ContextManager.stopSpan();
+        AbstractSpan localSpan = ContextManager.createLocalSpan("callback:" + uri);
+        localSpan.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        //stop local span:callback
+        ContextManager.stopSpan();
+        //stop local span:AsyncThread/execute
+        ContextManager.stopSpan();
+
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/SessionRequestImplIterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+
+/**
+ * Set local span false When  connect to the remote host  failed .
+ *
+ * @author liyuntao
+ */
+
+public class SessionRequestImplIterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/SetResponseInterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.http.HttpResponse;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+
+/**
+ * End exit span and create  a local span of  future/Callback.
+ *
+ * @author liyuntao
+ */
+
+public class SetResponseInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        if (null == allArguments[0]) {
+            return;
+        }
+        AbstractSpan span = ContextManager.activeSpan();
+        int statusCode = ((HttpResponse)allArguments[0]).getStatusLine().getStatusCode();
+        if (statusCode >= 400) {
+            span.errorOccurred();
+            Tags.STATUS_CODE.set(span, Integer.toString(statusCode));
+        }
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/StateInterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import java.net.MalformedURLException;
+import java.net.URL;
+import org.apache.http.client.methods.HttpRequestWrapper;
+import org.apache.skywalking.apm.agent.core.context.CarrierItem;
+import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * Create exit span of httpasyncclient.
+ *
+ * @author liyuntao
+ */
+
+public class StateInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        if (null == allArguments[0]) {
+            return;
+        }
+        HttpRequestWrapper httpRequest = (HttpRequestWrapper)allArguments[0];
+        String uri = httpRequest.getOriginal().getRequestLine().getUri();
+        AbstractSpan span = null;
+        final ContextCarrier contextCarrier = new ContextCarrier();
+        try {
+            URL url = new URL(httpRequest.getOriginal().getRequestLine().getUri());
+            String remotePeer = url.getHost() + ":" + url.getPort();
+            span = ContextManager.createExitSpan(url.getPath(), contextCarrier, remotePeer);
+        } catch (MalformedURLException e) {
+            throw e;
+        }
+        span.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT);
+        Tags.URL.set(span, uri);
+        Tags.HTTP.METHOD.set(span, httpRequest.getOriginal().getRequestLine().getMethod());
+        SpanLayer.asHttp(span);
+
+        CarrierItem next = contextCarrier.items();
+        while (next.hasNext()) {
+            next = next.next();
+            httpRequest.setHeader(next.getHeadKey(), next.getHeadValue());
+        }
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/SuccessInterceptor.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.lang.reflect.Method;
+import org.apache.http.nio.reactor.SessionRequest;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * Create a local sapn and passing ref accross thread by SessionRequest.
+ *
+ * @author liyuntao
+ */
+
+public class SuccessInterceptor implements InstanceMethodsAroundInterceptor {
+
+    @Override public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
+        SessionRequest request = (SessionRequest)allArguments[0];
+        AbstractSpan localSpan = ContextManager.createLocalSpan("AsyncThread/execute");
+        localSpan.setComponent(ComponentsDefine.HTTP_ASYNC_CLIENT).setLayer(SpanLayer.HTTP);
+        Object cacheValue = ((EnhancedInstance)request).getSkyWalkingDynamicField();
+        ContextManager.continued((ContextSnapshot)cacheValue);
+
+    }
+
+    @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Object ret) throws Throwable {
+        return ret;
+    }
+
+    @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+        Class<?>[] argumentsTypes, Throwable t) {
+        AbstractSpan activeSpan = ContextManager.activeSpan();
+        activeSpan.errorOccurred();
+        activeSpan.log(t);
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/AbstractNIOConnPoolInstrumentation.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link AbstractNIOConnPoolInstrumentation} presents that skywalking intercept {@link
+ * org.apache.http.nio.protocol.AbstractNIOConnPool #requestCompleted}.
+ *
+ * @author liyuntao
+ */
+
+public class AbstractNIOConnPoolInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.nio.pool.AbstractNIOConnPool";
+    private static final String START_LOCAL_SUCCESS_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.SuccessInterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("requestCompleted");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return START_LOCAL_SUCCESS_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/DefaultConnectingIOReactorInstrumentation.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link DefaultConnectingIOReactorInstrumentation} presents that skywalking intercepts {@link
+ * org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor#processEvent}
+ *
+ * @author liyuntao
+ */
+
+public class DefaultConnectingIOReactorInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor";
+    private static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.DefaultConnectingIOReactorIterceptor";
+    private static final String LOCAL_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.ConnectIterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("processEvent");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            },
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("connect");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return LOCAL_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/ExecuteInstrumentation.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link ExecuteInstrumentation} presents that skywalking intercepts {@link org.apache.http.impl.nio.client.CloseableHttpAsyncClient#execute}
+ *
+ * @author liyuntao
+ */
+
+public class ExecuteInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.impl.nio.client.CloseableHttpAsyncClient";
+    private static final String CONSUMER_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.HttpAsyncResponseConsumerInterceptor";
+    private static final String HOST_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.HttpHostInterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("execute").and(takesArgumentWithType(0, "org.apache.http.nio.protocol.HttpAsyncRequestProducer")).and(takesArguments(3));
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return CONSUMER_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            },
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("execute").and(takesArguments(4)).and(takesArgumentWithType(0, "org.apache.http.HttpHost"));
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return HOST_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/ProcessResponseInstrumentation.java

+/*
+ * 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.
+ *
+ */
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link ProcessResponseInstrumentation} presents that skywalking intercept {@link
+ * org.apache.http.nio.protocol.HttpAsyncRequestExecutor#processResponse,#connected} .
+ *
+ * @author liyuntao
+ */
+
+public class ProcessResponseInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.nio.protocol.HttpAsyncRequestExecutor";
+    private static final String END_EXIT_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.ProcessResponseInterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("processResponse");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return END_EXIT_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/SessionRequestImplInstrumentation.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link SessionRequestImplInstrumentation} presents that skywalking intercepts {@link
+ * org.apache.http.impl.nio.reactor.SessionRequestImpl#failed(final IOException exception)}
+ *
+ * @author liyuntao
+ */
+
+public class SessionRequestImplInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.impl.nio.reactor.SessionRequestImpl";
+    private static final String INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.SessionRequestImplIterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("failed");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/define/StateInstrumentation.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link StateInstrumentation} presents that skywalking intercept {@link org.apache.http.nio.protocol.HttpAsyncRequestExecutor$State#setRequest
+ * #setResponse} .
+ *
+ * @author liyuntao
+ */
+
+public class StateInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+    private static final String ENHANCE_CLASS = "org.apache.http.nio.protocol.HttpAsyncRequestExecutor$State";
+    private static final String START_EXIT_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.StateInterceptor";
+    private static final String END_EXIT_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.httpasyncclient.v4.SetResponseInterceptor";
+
+    @Override
+    public ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS);
+    }
+
+    @Override
+    protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("setRequest");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return START_EXIT_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            },
+            new InstanceMethodsInterceptPoint() {
+                @Override
+                public ElementMatcher<MethodDescription> getMethodsMatcher() {
+                    return named("setResponse");
+                }
+
+                @Override
+                public String getMethodsInterceptor() {
+                    return END_EXIT_INTERCEPT_CLASS;
+                }
+
+                @Override
+                public boolean isOverrideArgs() {
+                    return false;
+                }
+            }
+
+        };
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/main/resources/skywalking-plugin.def

+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.ExecuteInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.DefaultConnectingIOReactorInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.SessionRequestImplInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.AbstractNIOConnPoolInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.StateInstrumentation
+httpasyncclient-4.x=org.apache.skywalking.apm.plugin.httpasyncclient.v4.define.ProcessResponseInstrumentation
+
+

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/StateInterceptorTest.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.util.List;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolVersion;
+import org.apache.http.RequestLine;
+import org.apache.http.StatusLine;
+import org.apache.http.client.methods.HttpRequestWrapper;
+import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.context.util.KeyValuePair;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
+import org.apache.skywalking.apm.agent.test.helper.SpanHelper;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockRunnerDelegate(TracingSegmentRunner.class)
+@PrepareForTest(HttpHost.class)
+public class StateInterceptorTest {
+
+    @SegmentStoragePoint
+    private SegmentStorage segmentStorage;
+
+    @Rule
+    public AgentServiceRule agentServiceRule = new AgentServiceRule();
+
+    private StateInterceptor stateInterceptor;
+
+    private SetResponseInterceptor setResponseInterceptor;
+
+    private ProcessResponseInterceptor processResponseInterceptor;
+    @Mock
+    private HttpHost httpHost;
+    @Mock
+    private HttpRequestWrapper request;
+    @Mock
+    private HttpRequest httpRequest;
+    @Mock
+    private HttpResponse httpResponse;
+    @Mock
+    private StatusLine statusLine;
+
+    private Object[] allArguments;
+    private Class[] argumentsType;
+
+    @Mock
+    private EnhancedInstance enhancedInstance;
+
+    @Before
+    public void setUp() throws Exception {
+        ServiceManager.INSTANCE.boot();
+        stateInterceptor = new StateInterceptor();
+        setResponseInterceptor = new SetResponseInterceptor();
+        processResponseInterceptor = new ProcessResponseInterceptor();
+
+        PowerMockito.mock(HttpHost.class);
+        when(statusLine.getStatusCode()).thenReturn(200);
+        when(httpResponse.getStatusLine()).thenReturn(statusLine);
+        when(httpHost.getHostName()).thenReturn("127.0.0.1");
+        when(httpHost.getSchemeName()).thenReturn("http");
+        when(request.getOriginal()).thenReturn(httpRequest);
+        when(httpRequest.getRequestLine()).thenReturn(new RequestLine() {
+            @Override
+            public String getMethod() {
+                return "GET";
+            }
+
+            @Override
+            public ProtocolVersion getProtocolVersion() {
+                return new ProtocolVersion("http", 1, 1);
+            }
+
+            @Override
+            public String getUri() {
+                return "http://127.0.0.1:8080/test-web/httpasync";
+            }
+        });
+        when(httpHost.getPort()).thenReturn(8080);
+
+        allArguments = new Object[] {request};
+        argumentsType = new Class[] {request.getClass()};
+    }
+
+    @Test
+    public void testHttpClient() throws Throwable {
+        AbstractSpan span = ContextManager.createLocalSpan("httpasyncclient/HttpAsyncRequestExecutor:");
+        stateInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        stateInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        processResponseInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        processResponseInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        Assert.assertThat(segmentStorage.getTraceSegments().size(), is(1));
+        TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
+
+        List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
+        assertHttpSpan(spans.get(0));
+        verify(request, times(1)).setHeader(anyString(), anyString());
+    }
+
+    @Test
+    public void testStatusCodeNotEquals200() throws Throwable {
+        when(statusLine.getStatusCode()).thenReturn(500);
+        AbstractSpan span = ContextManager.createLocalSpan("httpasyncclient/HttpAsyncRequestExecutor:");
+        stateInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        stateInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        allArguments = new Object[] {httpResponse};
+        setResponseInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        setResponseInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        processResponseInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        processResponseInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+
+        Assert.assertThat(segmentStorage.getTraceSegments().size(), is(1));
+        TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
+        List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
+
+        assertThat(spans.size(), is(3));
+
+        List<KeyValuePair> tags = SpanHelper.getTags(spans.get(0));
+        assertThat(tags.size(), is(3));
+        assertThat(tags.get(2).getValue(), is("500"));
+
+        assertHttpSpan(spans.get(0));
+        assertThat(SpanHelper.getErrorOccurred(spans.get(0)), is(true));
+        verify(request, times(1)).setHeader(anyString(), anyString());
+    }
+
+    private void assertHttpSpan(AbstractTracingSpan span) {
+        assertThat(span.getOperationName(), is("/test-web/httpasync"));
+        assertThat(SpanHelper.getComponentId(span), is(26));
+        List<KeyValuePair> tags = SpanHelper.getTags(span);
+        assertThat(tags.get(0).getValue(), is("http://127.0.0.1:8080/test-web/httpasync"));
+        assertThat(tags.get(1).getValue(), is("GET"));
+        assertThat(span.isExit(), is(true));
+    }
+
+}

apm-sniffer/apm-sdk-plugin/httpasyncclient-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/httpasyncclient/v4/TestException.java

+/*
+ * 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.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.httpasyncclient.v4;
+
+import java.util.List;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolVersion;
+import org.apache.http.RequestLine;
+import org.apache.http.StatusLine;
+import org.apache.http.client.methods.HttpRequestWrapper;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.LogDataEntity;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
+import org.apache.skywalking.apm.agent.test.helper.SpanHelper;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+
+import static junit.framework.TestCase.assertNotNull;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @auther lytscu
+ */
+@RunWith(PowerMockRunner.class)
+@PowerMockRunnerDelegate(TracingSegmentRunner.class)
+@PrepareForTest(HttpHost.class)
+public class TestException {
+    @SegmentStoragePoint
+    private SegmentStorage segmentStorage;
+
+    @Rule
+    public AgentServiceRule agentServiceRule = new AgentServiceRule();
+
+    private StateInterceptor stateInterceptor;
+
+    private SetResponseInterceptor setResponseInterceptor;
+
+    private ProcessResponseInterceptor processResponseInterceptor;
+    @Mock
+    private HttpHost httpHost;
+    @Mock
+    private HttpRequestWrapper request;
+    @Mock
+    private HttpRequest httpRequest;
+    @Mock
+    private HttpResponse httpResponse;
+    @Mock
+    private StatusLine statusLine;
+
+    private Object[] allArguments, setResponseInterceptorArguments;
+    private Class[] argumentsType;
+
+    @Mock
+    private EnhancedInstance enhancedInstance;
+
+    @Before
+    public void setUp() throws Exception {
+        ServiceManager.INSTANCE.boot();
+        stateInterceptor = new StateInterceptor();
+        setResponseInterceptor = new SetResponseInterceptor();
+        processResponseInterceptor = new ProcessResponseInterceptor();
+        PowerMockito.mock(HttpHost.class);
+        when(statusLine.getStatusCode()).thenReturn(200);
+        when(httpResponse.getStatusLine()).thenReturn(statusLine);
+        when(httpHost.getHostName()).thenReturn("127.0.0.1");
+        when(httpHost.getSchemeName()).thenReturn("http");
+        when(request.getOriginal()).thenReturn(httpRequest);
+        when(httpRequest.getRequestLine()).thenReturn(new RequestLine() {
+            @Override
+            public String getMethod() {
+                return "GET";
+            }
+
+            @Override
+            public ProtocolVersion getProtocolVersion() {
+                return new ProtocolVersion("http", 1, 1);
+            }
+
+            @Override
+            public String getUri() {
+                return "http://127.0.0.1:8080/test-web/httpasync";
+            }
+        });
+        when(httpHost.getPort()).thenReturn(8080);
+
+        allArguments = new Object[] {request};
+        setResponseInterceptorArguments = new Object[] {httpResponse};
+        argumentsType = new Class[] {request.getClass()};
+    }
+
+    @Test
+    public void testHttpClientWithException() throws Throwable {
+        AbstractSpan localSpan = ContextManager.createLocalSpan("httpasyncclient/HttpAsyncRequestExecutor:");
+        stateInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        stateInterceptor.handleMethodException(enhancedInstance, null, allArguments, argumentsType, new RuntimeException("testException"));
+        processResponseInterceptor.beforeMethod(enhancedInstance, null, allArguments, argumentsType, null);
+        processResponseInterceptor.afterMethod(enhancedInstance, null, allArguments, argumentsType, httpResponse);
+        Assert.assertThat(segmentStorage.getTraceSegments().size(), is(1));
+        TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
+        List<AbstractTracingSpan> spans = SegmentHelper.getSpans(traceSegment);
+
+        assertThat(spans.size(), is(3));
+        AbstractTracingSpan span = spans.get(0);
+        assertThat(SpanHelper.getErrorOccurred(span), is(true));
+        assertHttpSpanErrorLog(SpanHelper.getLogs(span));
+        verify(request, times(1)).setHeader(anyString(), anyString());
+
+    }
+
+    private void assertHttpSpanErrorLog(List<LogDataEntity> logs) {
+        assertThat(logs.size(), is(1));
+        LogDataEntity logData = logs.get(0);
+        Assert.assertThat(logData.getLogs().size(), is(4));
+        Assert.assertThat(logData.getLogs().get(0).getValue(), CoreMatchers.<Object>is("error"));
+        Assert.assertThat(logData.getLogs().get(1).getValue(), CoreMatchers.<Object>is(RuntimeException.class.getName()));
+        Assert.assertThat(logData.getLogs().get(2).getValue(), is("testException"));
+        assertNotNull(logData.getLogs().get(3).getValue());
+    }
+}

apm-sniffer/apm-sdk-plugin/pom.xml

@@ -52,6 +52,7 @@
         <module>rocketMQ-4.x-plugin</module>
         <module>elastic-job-2.x-plugin</module>
         <module>mongodb-2.x-plugin</module>
+        <module>httpasyncclient-4.x-plugin</module>
     </modules>
     <packaging>pom</packaging>
 

apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptor.java

@@ -20,12 +20,12 @@
 package org.apache.skywalking.apm.plugin.rocketMQ.v4;
 
 import java.lang.reflect.Method;
-import org.apache.rocketmq.client.impl.CommunicationMode;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
 import org.apache.skywalking.apm.agent.core.context.CarrierItem;
 import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
 import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.context.tag.Tags;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
 import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
@@ -60,10 +60,9 @@ public class MessageSendInterceptor implements InstanceMethodsAroundInterceptor
         String namingServiceAddress = String.valueOf(objInst.getSkyWalkingDynamicField());
         AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), contextCarrier, namingServiceAddress);
         span.setComponent(ComponentsDefine.ROCKET_MQ);
+        Tags.MQ_BROKER.set(span, (String)allArguments[0]);
+        Tags.MQ_TOPIC.set(span, message.getTopic());
         SpanLayer.asMQ(span);
-        span.tag("brokerName", (String)allArguments[1]);
-        span.tag("tags", message.getTags());
-        span.tag("communication.mode", ((CommunicationMode)allArguments[5]).name());
 
         SendMessageRequestHeader requestHeader = (SendMessageRequestHeader)allArguments[3];
         StringBuilder properties = new StringBuilder(requestHeader.getProperties());

apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptorTest.java

@@ -110,8 +110,7 @@ public class MessageSendInterceptorTest {
 
         SpanAssert.assertLayer(mqSpan, SpanLayer.MQ);
         SpanAssert.assertComponent(mqSpan, ComponentsDefine.ROCKET_MQ);
-        SpanAssert.assertTag(mqSpan, 0, "test");
-        SpanAssert.assertTag(mqSpan, 1, "TagA");
+        SpanAssert.assertTag(mqSpan, 0, "127.0.0.1");
         verify(messageRequestHeader, times(1)).setProperties(anyString());
         verify(callBack, times(1)).setSkyWalkingDynamicField(Matchers.any());
     }
@@ -130,8 +129,7 @@ public class MessageSendInterceptorTest {
 
         SpanAssert.assertLayer(mqSpan, SpanLayer.MQ);
         SpanAssert.assertComponent(mqSpan, ComponentsDefine.ROCKET_MQ);
-        SpanAssert.assertTag(mqSpan, 0, "test");
-        SpanAssert.assertTag(mqSpan, 1, "TagA");
+        SpanAssert.assertTag(mqSpan, 0, "127.0.0.1");
         verify(messageRequestHeader, times(1)).setProperties(anyString());
     }