Merge branch 'feature/collector-modelization' of https://github.com/OpenSkywalking/skywalking into feature/collector-modelization

# Conflicts:
#	apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractLocalAsyncWorkerProvider.java
#	apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractRemoteWorkerProvider.java
#	apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractWorkerProvider.java

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ApplicationCacheGuavaService.java

@@ -35,7 +35,7 @@ public class ApplicationCacheGuavaService implements ApplicationCacheService {
 
     private final Logger logger = LoggerFactory.getLogger(ApplicationCacheGuavaService.class);
 
-    private final Cache<String, Integer> CODE_CACHE = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(1000).build();
+    private final Cache<String, Integer> codeCache = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(1000).build();
 
     private final DAOService daoService;
 
@@ -48,7 +48,7 @@ public class ApplicationCacheGuavaService implements ApplicationCacheService {
 
         int applicationId = 0;
         try {
-            applicationId = CODE_CACHE.get(applicationCode, () -> dao.getApplicationId(applicationCode));
+            applicationId = codeCache.get(applicationCode, () -> dao.getApplicationId(applicationCode));
         } catch (Throwable e) {
             logger.error(e.getMessage(), e);
         }
@@ -56,20 +56,20 @@ public class ApplicationCacheGuavaService implements ApplicationCacheService {
         if (applicationId == 0) {
             applicationId = dao.getApplicationId(applicationCode);
             if (applicationId != 0) {
-                CODE_CACHE.put(applicationCode, applicationId);
+                codeCache.put(applicationCode, applicationId);
             }
         }
         return applicationId;
     }
 
-    private final Cache<Integer, String> ID_CACHE = CacheBuilder.newBuilder().maximumSize(1000).build();
+    private final Cache<Integer, String> idCache = CacheBuilder.newBuilder().maximumSize(1000).build();
 
     public String get(int applicationId) {
         IApplicationCacheDAO dao = (IApplicationCacheDAO)daoService.get(IApplicationCacheDAO.class);
 
         String applicationCode = Const.EMPTY_STRING;
         try {
-            applicationCode = ID_CACHE.get(applicationId, () -> dao.getApplicationCode(applicationId));
+            applicationCode = idCache.get(applicationId, () -> dao.getApplicationCode(applicationId));
         } catch (Throwable e) {
             logger.error(e.getMessage(), e);
         }
@@ -77,7 +77,7 @@ public class ApplicationCacheGuavaService implements ApplicationCacheService {
         if (StringUtils.isEmpty(applicationCode)) {
             applicationCode = dao.getApplicationCode(applicationId);
             if (StringUtils.isNotEmpty(applicationCode)) {
-                CODE_CACHE.put(applicationCode, applicationId);
+                codeCache.put(applicationCode, applicationId);
             }
         }
         return applicationCode;

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/InstanceCacheGuavaService.java

@@ -33,7 +33,7 @@ public class InstanceCacheGuavaService implements InstanceCacheService {
 
     private final Logger logger = LoggerFactory.getLogger(InstanceCacheGuavaService.class);
 
-    private final Cache<Integer, Integer> INSTANCE_CACHE = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(5000).build();
+    private final Cache<Integer, Integer> integerCache = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(5000).build();
 
     private final DAOService daoService;
 
@@ -46,7 +46,7 @@ public class InstanceCacheGuavaService implements InstanceCacheService {
 
         int applicationId = 0;
         try {
-            applicationId = INSTANCE_CACHE.get(applicationInstanceId, () -> dao.getApplicationId(applicationInstanceId));
+            applicationId = integerCache.get(applicationInstanceId, () -> dao.getApplicationId(applicationInstanceId));
         } catch (Throwable e) {
             logger.error(e.getMessage(), e);
         }
@@ -54,7 +54,7 @@ public class InstanceCacheGuavaService implements InstanceCacheService {
         if (applicationId == 0) {
             applicationId = dao.getApplicationId(applicationInstanceId);
             if (applicationId != 0) {
-                INSTANCE_CACHE.put(applicationInstanceId, applicationId);
+                integerCache.put(applicationInstanceId, applicationId);
             }
         }
         return applicationId;

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceIdCacheGuavaService.java

@@ -34,7 +34,7 @@ public class ServiceIdCacheGuavaService implements ServiceIdCacheService {
 
     private final Logger logger = LoggerFactory.getLogger(ServiceIdCacheGuavaService.class);
 
-    private final Cache<String, Integer> SERVICE_CACHE = CacheBuilder.newBuilder().maximumSize(1000).build();
+    private final Cache<String, Integer> serviceIdCache = CacheBuilder.newBuilder().maximumSize(1000).build();
 
     private final DAOService daoService;
 
@@ -47,7 +47,7 @@ public class ServiceIdCacheGuavaService implements ServiceIdCacheService {
 
         int serviceId = 0;
         try {
-            serviceId = SERVICE_CACHE.get(applicationId + Const.ID_SPLIT + serviceName, () -> dao.getServiceId(applicationId, serviceName));
+            serviceId = serviceIdCache.get(applicationId + Const.ID_SPLIT + serviceName, () -> dao.getServiceId(applicationId, serviceName));
         } catch (Throwable e) {
             logger.error(e.getMessage(), e);
         }
@@ -55,7 +55,7 @@ public class ServiceIdCacheGuavaService implements ServiceIdCacheService {
         if (serviceId == 0) {
             serviceId = dao.getServiceId(applicationId, serviceName);
             if (serviceId != 0) {
-                SERVICE_CACHE.put(applicationId + Const.ID_SPLIT + serviceName, serviceId);
+                serviceIdCache.put(applicationId + Const.ID_SPLIT + serviceName, serviceId);
             }
         }
         return serviceId;

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceNameCacheGuavaService.java

@@ -35,7 +35,7 @@ public class ServiceNameCacheGuavaService implements ServiceNameCacheService {
 
     private final Logger logger = LoggerFactory.getLogger(ServiceNameCacheGuavaService.class);
 
-    private final Cache<Integer, String> CACHE = CacheBuilder.newBuilder().maximumSize(10000).build();
+    private final Cache<Integer, String> serviceNameCache = CacheBuilder.newBuilder().maximumSize(10000).build();
 
     private final DAOService daoService;
 
@@ -48,7 +48,7 @@ public class ServiceNameCacheGuavaService implements ServiceNameCacheService {
 
         String serviceName = Const.EMPTY_STRING;
         try {
-            serviceName = CACHE.get(serviceId, () -> dao.getServiceName(serviceId));
+            serviceName = serviceNameCache.get(serviceId, () -> dao.getServiceName(serviceId));
         } catch (Throwable e) {
             logger.error(e.getMessage(), e);
         }
@@ -56,7 +56,7 @@ public class ServiceNameCacheGuavaService implements ServiceNameCacheService {
         if (StringUtils.isEmpty(serviceName)) {
             serviceName = dao.getServiceName(serviceId);
             if (StringUtils.isNotEmpty(serviceName)) {
-                CACHE.put(serviceId, serviceName);
+                serviceNameCache.put(serviceId, serviceName);
             }
         }
 

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Graph.java

@@ -25,21 +25,26 @@ import java.util.concurrent.ConcurrentHashMap;
  */
 public final class Graph<INPUT> {
     private int id;
-    private Node startNode;
+    private WayToNode entryWay;
     private ConcurrentHashMap<Integer, Node> nodeIndex = new ConcurrentHashMap<>();
 
     Graph(int id) {
         this.id = id;
     }
 
-    public void start(INPUT INPUT) {
-        startNode.execute(INPUT);
+    public void start(INPUT input) {
+        entryWay.in(input);
     }
 
     public <OUTPUT> Node<INPUT, OUTPUT> addNode(NodeProcessor<INPUT, OUTPUT> nodeProcessor) {
+        return addNode(new DirectWay(nodeProcessor));
+    }
+
+    public <OUTPUT> Node<INPUT, OUTPUT> addNode(WayToNode<INPUT, OUTPUT> entryWay) {
         synchronized (this) {
-            startNode = new Node(this, nodeProcessor);
-            return startNode;
+            this.entryWay = entryWay;
+            this.entryWay.buildDestination(this);
+            return entryWay.getDestination();
         }
     }
 
@@ -48,7 +53,7 @@ public final class Graph<INPUT> {
         if (nodeIndex.containsKey(nodeId)) {
             throw new PotentialCyclicGraphException("handler="
                 + node.getHandler().getClass().getName()
-                + " already exists in graph[" + id + "】");
+                + " already exists in graph[" + id + "]");
         }
         nodeIndex.put(nodeId, node);
     }

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/GraphManager.java

@@ -35,7 +35,7 @@ public enum GraphManager {
      * @param graphId represents a graph, which is used for finding it.
      * @return
      */
-    public synchronized <Input> Graph<Input> createIfAbsent(int graphId, Class<Input> input) {
+    public synchronized <INPUT> Graph<INPUT> createIfAbsent(int graphId, Class<INPUT> input) {
         if (!allGraphs.containsKey(graphId)) {
             Graph graph = new Graph(graphId);
             allGraphs.put(graphId, graph);

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/GraphNodeFinder.java

@@ -35,10 +35,10 @@ public class GraphNodeFinder {
      *
      * @param handlerId of specific node in graph.
      * @param outputClass of the found node
-     * @param <NODE_OUTPUT> type of given output class
+     * @param <NODEOUTPUT> type of given output class
      * @return Node instance.
      */
-    public <NODE_OUTPUT> Node<?, NODE_OUTPUT> findNode(int handlerId, Class<NODE_OUTPUT> outputClass) {
+    public <NODEOUTPUT> Node<?, NODEOUTPUT> findNode(int handlerId, Class<NODEOUTPUT> outputClass) {
         ConcurrentHashMap<Integer, Node> graphNodeIndex = graph.getNodeIndex();
         Node node = graphNodeIndex.get(handlerId);
         if (node == null) {

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Next.java

@@ -42,9 +42,9 @@ public class Next<INPUT> implements Executor<INPUT> {
     /**
      * Drive to the next nodes
      *
-     * @param INPUT
+     * @param input
      */
-    @Override public void execute(INPUT INPUT) {
-        ways.forEach(way -> way.in(INPUT));
+    @Override public void execute(INPUT input) {
+        ways.forEach(way -> way.in(input));
     }
 }

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/Node.java

@@ -47,8 +47,8 @@ public final class Node<INPUT, OUTPUT> {
         }
     }
 
-    final void execute(INPUT INPUT) {
-        nodeProcessor.process(INPUT, next);
+    final void execute(INPUT input) {
+        nodeProcessor.process(input, next);
     }
 
     NodeProcessor getHandler() {

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/NodeProcessor.java

@@ -21,7 +21,7 @@ package org.skywalking.apm.collector.core.graph;
 /**
  * @author peng-yongsheng, wu-sheng
  */
-public interface NodeProcessor<INPUT, OUTPUT> {
+public interface NodeProcessor<input, output> {
     /**
      * The unique id in the certain graph.
      *
@@ -29,5 +29,5 @@ public interface NodeProcessor<INPUT, OUTPUT> {
      */
     int id();
 
-    void process(INPUT INPUT, Next<OUTPUT> next);
+    void process(input input, Next<output> next);
 }

apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/graph/WayToNode.java

@@ -33,10 +33,10 @@ public abstract class WayToNode<INPUT, OUTPUT> {
         destination = new Node(graph, destinationHandler);
     }
 
-    protected abstract void in(INPUT INPUT);
+    protected abstract void in(INPUT input);
 
-    protected void out(INPUT INPUT) {
-        destination.execute(INPUT);
+    protected void out(INPUT input) {
+        destination.execute(input);
     }
 
     Node getDestination() {

apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/GraphManagerTest.java

@@ -31,7 +31,7 @@ import org.junit.Test;
 public class GraphManagerTest {
     private static PrintStream OUT_REF;
     private ByteArrayOutputStream outputStream;
-    private static String lineSeparator = System.lineSeparator();
+    private static String LINE_SEPARATE = System.lineSeparator();
 
     @Before
     public void initAndHoldOut() {
@@ -54,8 +54,8 @@ public class GraphManagerTest {
         testGraph.start("Input String");
 
         String output = outputStream.toString();
-        String expected = "Node1 process: s=Input String" + lineSeparator +
-            "Node2 process: s=Input String" + lineSeparator;
+        String expected = "Node1 process: s=Input String" + LINE_SEPARATE +
+            "Node2 process: s=Input String" + LINE_SEPARATE;
 
         Assert.assertEquals(expected, output);
     }
@@ -68,9 +68,9 @@ public class GraphManagerTest {
         graph.start("Input String");
 
         String output = outputStream.toString();
-        String expected = "Node1 process: s=Input String" + lineSeparator +
-            "Node2 process: s=Input String" + lineSeparator +
-            "Node4 process: int=123" + lineSeparator;
+        String expected = "Node1 process: s=Input String" + LINE_SEPARATE +
+            "Node2 process: s=Input String" + LINE_SEPARATE +
+            "Node4 process: int=123" + LINE_SEPARATE;
 
         Assert.assertEquals(expected, output);
     }
@@ -92,7 +92,7 @@ public class GraphManagerTest {
         next.execute(123);
         String output = outputStream.toString();
         String expected =
-            "Node4 process: int=123" + lineSeparator;
+            "Node4 process: int=123" + LINE_SEPARATE;
 
         Assert.assertEquals(expected, output);
     }
@@ -118,18 +118,32 @@ public class GraphManagerTest {
     public void testDeadEndWay() {
         Graph<String> graph = GraphManager.INSTANCE.createIfAbsent(7, String.class);
         graph.addNode(new Node1Processor()).addNext(new WayToNode<String, Integer>(new Node2Processor()) {
-            @Override protected void in(String INPUT) {
+            @Override protected void in(String input) {
                 //don't call `out(intput)`;
             }
         });
 
         graph.start("Input String");
         String output = outputStream.toString();
-        String expected = "Node1 process: s=Input String" + lineSeparator;
+        String expected = "Node1 process: s=Input String" + LINE_SEPARATE;
 
         Assert.assertEquals(expected, output);
     }
 
+    @Test
+    public void testEntryWay() {
+        Graph<String> graph = GraphManager.INSTANCE.createIfAbsent(8, String.class);
+        graph.addNode(new WayToNode<String, String>(new Node1Processor()) {
+            @Override protected void in(String input) {
+                //don't call `out(intput)`;
+            }
+        }).addNext(new Node2Processor());
+
+        graph.start("Input String");
+
+        Assert.assertEquals("", outputStream.toString());
+    }
+
     @After
     public void tearDown() {
         GraphManager.INSTANCE.reset();

apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node1Processor.java

@@ -21,13 +21,13 @@ package org.skywalking.apm.collector.core.graph;
 /**
  * @author wusheng
  */
-public class Node1Processor implements NodeProcessor<String, String> {
+public class Node1Processor extends OutputProcessor implements NodeProcessor<String, String> {
     @Override public int id() {
         return 1;
     }
 
     @Override public void process(String s, Next<String> next) {
-        System.out.println("Node1 process: s=" + s);
+        outstream().println("Node1 process: s=" + s);
         next.execute(s);
     }
 }

apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node2Processor.java

@@ -21,13 +21,13 @@ package org.skywalking.apm.collector.core.graph;
 /**
  * @author wusheng
  */
-public class Node2Processor implements NodeProcessor<String, Integer> {
+public class Node2Processor extends OutputProcessor implements NodeProcessor<String, Integer> {
     @Override public int id() {
         return 2;
     }
 
     @Override public void process(String s, Next<Integer> next) {
-        System.out.println("Node2 process: s=" + s);
+        outstream().println("Node2 process: s=" + s);
         next.execute(123);
     }
 }

apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node3Processor.java

@@ -21,14 +21,14 @@ package org.skywalking.apm.collector.core.graph;
 /**
  * @author wusheng
  */
-public class Node3Processor implements NodeProcessor<Long, Long> {
+public class Node3Processor extends OutputProcessor  implements NodeProcessor<Long, Long> {
     @Override public int id() {
         return 3;
     }
 
     @Override
     public void process(Long aLong, Next<Long> next) {
-        System.out.println("Node3 process: long=" + aLong);
+        outstream().println("Node3 process: long=" + aLong);
         next.execute(aLong);
     }
 }

apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/Node4Processor.java

@@ -21,14 +21,14 @@ package org.skywalking.apm.collector.core.graph;
 /**
  * @author wusheng
  */
-public class Node4Processor implements NodeProcessor<Integer, Long> {
+public class Node4Processor extends OutputProcessor implements NodeProcessor<Integer, Long> {
     @Override public int id() {
         return 4;
     }
 
     @Override
     public void process(Integer in, Next<Long> next) {
-        System.out.println("Node4 process: int=" + in);
+        outstream().println("Node4 process: int=" + in);
         next.execute(new Long(in.intValue()));
     }
 }

apm-collector/apm-collector-core/src/test/java/org/skywalking/apm/collector/core/graph/OutputProcessor.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed 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 repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.core.graph;
+
+import java.io.PrintStream;
+
+/**
+ * @author wu-sheng
+ */
+public class OutputProcessor {
+    protected PrintStream outstream() {
+        return System.out;
+    }
+}

apm-collector/apm-collector-remote/collector-remote-kafka-provider/pom.xml

@@ -30,7 +30,6 @@
     <artifactId>collector-remote-kafka-provider</artifactId>
     <packaging>jar</packaging>
 
-<<<<<<< HEAD
     <dependencies>
         <dependency>
             <groupId>org.skywalking</groupId>
@@ -39,6 +38,3 @@
         </dependency>
     </dependencies>
 </project>
-=======
-</project>
->>>>>>> 406c4f52d8251eb81aa868ec38f17789e18e0dc0

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractLocalAsyncWorkerProvider.java

@@ -28,7 +28,7 @@ import org.skywalking.apm.collector.storage.service.DAOService;
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractLocalAsyncWorkerProvider<INPUT extends Data, OUTPUT extends Data, WorkerType extends AbstractLocalAsyncWorker<INPUT, OUTPUT> & QueueExecutor<INPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WorkerType> {
+public abstract class AbstractLocalAsyncWorkerProvider<INPUT extends Data, OUTPUT extends Data, WORKER_TYPE extends AbstractLocalAsyncWorker<INPUT, OUTPUT> & QueueExecutor<INPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
 
     public abstract int queueSize();
 
@@ -42,7 +42,11 @@ public abstract class AbstractLocalAsyncWorkerProvider<INPUT extends Data, OUTPU
 
     @Override
     final public WorkerRef create(WorkerCreateListener workerCreateListener) throws ProviderNotFoundException {
+<<<<<<< HEAD
         WorkerType localAsyncWorker = workerInstance(getDaoService(), getCacheServiceManager());
+=======
+        WORKER_TYPE localAsyncWorker = workerInstance(daoService);
+>>>>>>> 0c17906c3c1c41752e1ec38b37d9e0dec22503ca
         workerCreateListener.addWorker(localAsyncWorker);
         QueueEventHandler<INPUT> queueEventHandler = queueCreatorService.create(queueSize(), localAsyncWorker);
         return new LocalAsyncWorkerRef<>(localAsyncWorker, queueEventHandler);

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractRemoteWorkerProvider.java

@@ -31,7 +31,7 @@ import org.skywalking.apm.collector.storage.service.DAOService;
  * @author peng-yongsheng
  * @since v3.0-2017
  */
-public abstract class AbstractRemoteWorkerProvider<INPUT extends Data, OUTPUT extends Data, WorkerType extends AbstractRemoteWorker<INPUT, OUTPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WorkerType> {
+public abstract class AbstractRemoteWorkerProvider<INPUT extends Data, OUTPUT extends Data, WORKER_TYPE extends AbstractRemoteWorker<INPUT, OUTPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
 
     private final RemoteClientService remoteClientService;
 
@@ -49,7 +49,11 @@ public abstract class AbstractRemoteWorkerProvider<INPUT extends Data, OUTPUT ex
      * worker instance, when the worker provider not find then Throw this Exception.
      */
     @Override final public WorkerRef create(WorkerCreateListener workerCreateListener) {
+<<<<<<< HEAD
         WorkerType remoteWorker = workerInstance(getDaoService(), getCacheServiceManager());
+=======
+        WORKER_TYPE remoteWorker = workerInstance(daoService);
+>>>>>>> 0c17906c3c1c41752e1ec38b37d9e0dec22503ca
         workerCreateListener.addWorker(remoteWorker);
         RemoteWorkerRef<INPUT, OUTPUT> workerRef = new RemoteWorkerRef<>(remoteWorker);
         return workerRef;

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractWorker.java

@@ -59,10 +59,10 @@ public abstract class AbstractWorker<INPUT extends Data, OUTPUT extends Data> im
      */
     protected abstract void onWork(INPUT message) throws WorkerException;
 
-    @Override public final void process(INPUT INPUT, Next<OUTPUT> next) {
+    @Override public final void process(INPUT input, Next<OUTPUT> next) {
         this.next = next;
         try {
-            onWork(INPUT);
+            onWork(input);
         } catch (WorkerException e) {
             logger.error(e.getMessage(), e);
         }

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/AbstractWorkerProvider.java

@@ -25,6 +25,7 @@ import org.skywalking.apm.collector.storage.service.DAOService;
 /**
  * @author peng-yongsheng
  */
+<<<<<<< HEAD
 public abstract class AbstractWorkerProvider<INPUT extends Data, OUTPUT extends Data, WorkerType extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
 
     private final DAOService daoService;
@@ -44,4 +45,8 @@ public abstract class AbstractWorkerProvider<INPUT extends Data, OUTPUT extends
     }
 
     public abstract WorkerType workerInstance(DAOService daoService, CacheServiceManager cacheServiceManager);
+=======
+public abstract class AbstractWorkerProvider<INPUT extends Data, OUTPUT extends Data, WORKER_TYPE extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
+    public abstract WORKER_TYPE workerInstance(DAOService daoService);
+>>>>>>> 0c17906c3c1c41752e1ec38b37d9e0dec22503ca
 }

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/LocalAsyncWorkerRef.java

@@ -35,11 +35,11 @@ public class LocalAsyncWorkerRef<INPUT extends Data, OUTPUT extends Data> extend
         this.queueEventHandler = queueEventHandler;
     }
 
-    @Override protected void in(INPUT INPUT) {
-        queueEventHandler.tell(INPUT);
+    @Override protected void in(INPUT input) {
+        queueEventHandler.tell(input);
     }
 
-    @Override protected void out(INPUT INPUT) {
-        super.out(INPUT);
+    @Override protected void out(INPUT input) {
+        super.out(input);
     }
 }

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/RemoteWorkerRef.java

@@ -61,8 +61,8 @@ public class RemoteWorkerRef<INPUT extends Data, OUTPUT extends Data> extends Wo
         }
     }
 
-    @Override protected void out(INPUT INPUT) {
-        super.out(INPUT);
+    @Override protected void out(INPUT input) {
+        super.out(input);
     }
 
     private Boolean isAcrossJVM() {

checkStyle.xml

@@ -81,11 +81,11 @@
         <module name="MemberName"/>
         <!--Validates identifiers for class type parameters-->
         <module name="ClassTypeParameterName">
-            <property name="format" value="^[A-Z0-9]*$"/>
+            <property name="format" value="(^[A-Z][A-Z0-9]*(_[A-Z0-9]+)*$)"/>
         </module>
         <!--Validates identifiers for method type parameters-->
         <module name="MethodTypeParameterName">
-            <property name="format" value="^[A-Z0-9]*$"/>
+            <property name="format" value="(^[A-Z][A-Z0-9]*(_[A-Z0-9]+)*$)"/>
         </module>
         <module name="PackageName"/>
         <module name="ParameterName"/>