Collector cluster mode stream error (#899)

* Get the MQ server address id from reference.

* Fixed the remote exception bug when running cluster mode collector.
How: Register remote data entity into remote module.

apm-collector/apm-collector-agent/agent-grpc/agent-grpc-provider/src/test/java/org/apache/skywalking/apm/collector/agent/grpc/provider/handler/mock/RegisterMock.java

@@ -99,7 +99,7 @@ class RegisterMock {
         serviceNameElement.setSrcSpanType(SpanType.Exit);
         serviceNameCollection.addElements(serviceNameElement);
 
-        registerServiceName(serviceNameCollection);
+//        registerServiceName(serviceNameCollection);
 
         heartBeatScheduled(instanceMapping.getApplicationInstanceId());
     }
@@ -144,7 +144,7 @@ class RegisterMock {
         serviceNameElement.setSrcSpanType(SpanType.Entry);
         serviceNameCollection.addElements(serviceNameElement);
 
-        registerServiceName(serviceNameCollection);
+//        registerServiceName(serviceNameCollection);
 
         heartBeatScheduled(instanceMapping.getApplicationInstanceId());
     }

apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/AnalysisAlarmModuleProvider.java

@@ -34,7 +34,14 @@ import org.apache.skywalking.apm.collector.core.module.Module;
 import org.apache.skywalking.apm.collector.core.module.ModuleProvider;
 import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
 import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.table.alarm.ApplicationAlarm;
+import org.apache.skywalking.apm.collector.storage.table.alarm.ApplicationAlarmList;
+import org.apache.skywalking.apm.collector.storage.table.alarm.InstanceAlarm;
+import org.apache.skywalking.apm.collector.storage.table.alarm.InstanceAlarmList;
+import org.apache.skywalking.apm.collector.storage.table.alarm.ServiceAlarm;
+import org.apache.skywalking.apm.collector.storage.table.alarm.ServiceAlarmList;
 
 /**
  * @author peng-yongsheng
@@ -74,6 +81,8 @@ public class AnalysisAlarmModuleProvider extends ModuleProvider {
         ApplicationReferenceMetricAlarmGraph applicationReferenceMetricAlarmGraph = new ApplicationReferenceMetricAlarmGraph(getManager(), workerCreateListener);
         applicationReferenceMetricAlarmGraph.create();
 
+        registerRemoteData();
+
         PersistenceTimer persistenceTimer = new PersistenceTimer(AnalysisAlarmModule.NAME);
         persistenceTimer.start(getManager(), workerCreateListener.getPersistenceWorkers());
     }
@@ -85,4 +94,15 @@ public class AnalysisAlarmModuleProvider extends ModuleProvider {
     @Override public String[] requiredModules() {
         return new String[] {RemoteModule.NAME, AnalysisMetricModule.NAME, ConfigurationModule.NAME, StorageModule.NAME};
     }
+
+    private void registerRemoteData() {
+        RemoteDataRegisterService remoteDataRegisterService = getManager().find(RemoteModule.NAME).getService(RemoteDataRegisterService.class);
+        remoteDataRegisterService.register(ApplicationAlarm.class, new ApplicationAlarm.InstanceCreator());
+        remoteDataRegisterService.register(ApplicationAlarmList.class, new ApplicationAlarmList.InstanceCreator());
+        remoteDataRegisterService.register(InstanceAlarm.class, new InstanceAlarm.InstanceCreator());
+        remoteDataRegisterService.register(InstanceAlarmList.class, new InstanceAlarmList.InstanceCreator());
+        remoteDataRegisterService.register(ServiceAlarm.class, new ServiceAlarm.InstanceCreator());
+        remoteDataRegisterService.register(ServiceAlarmList.class, new ServiceAlarmList.InstanceCreator());
+
+    }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java

@@ -47,6 +47,17 @@ import org.apache.skywalking.apm.collector.analysis.worker.timer.PersistenceTime
 import org.apache.skywalking.apm.collector.core.module.Module;
 import org.apache.skywalking.apm.collector.core.module.ModuleProvider;
 import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException;
+import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
+import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent;
+import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping;
+import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric;
+import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMapping;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
+import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
+import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
 
 /**
  * @author peng-yongsheng
@@ -74,6 +85,8 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
 
         graphCreate(workerCreateListener);
 
+        registerRemoteData();
+
         PersistenceTimer persistenceTimer = new PersistenceTimer(AnalysisMetricModule.NAME);
         persistenceTimer.start(getManager(), workerCreateListener.getPersistenceWorkers());
     }
@@ -133,4 +146,17 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
         InstanceHeartBeatPersistenceGraph instanceHeartBeatPersistenceGraph = new InstanceHeartBeatPersistenceGraph(getManager(), workerCreateListener);
         instanceHeartBeatPersistenceGraph.create();
     }
+
+    private void registerRemoteData() {
+        RemoteDataRegisterService remoteDataRegisterService = getManager().find(RemoteModule.NAME).getService(RemoteDataRegisterService.class);
+        remoteDataRegisterService.register(ApplicationComponent.class, new ApplicationComponent.InstanceCreator());
+        remoteDataRegisterService.register(ApplicationMapping.class, new ApplicationMapping.InstanceCreator());
+        remoteDataRegisterService.register(ApplicationMetric.class, new ApplicationMetric.InstanceCreator());
+        remoteDataRegisterService.register(ApplicationReferenceMetric.class, new ApplicationReferenceMetric.InstanceCreator());
+        remoteDataRegisterService.register(InstanceMapping.class, new InstanceMapping.InstanceCreator());
+        remoteDataRegisterService.register(InstanceMetric.class, new InstanceMetric.InstanceCreator());
+        remoteDataRegisterService.register(InstanceReferenceMetric.class, new InstanceReferenceMetric.InstanceCreator());
+        remoteDataRegisterService.register(ServiceMetric.class, new ServiceMetric.InstanceCreator());
+        remoteDataRegisterService.register(ServiceReferenceMetric.class, new ServiceReferenceMetric.InstanceCreator());
+    }
 }

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

@@ -39,7 +39,12 @@ import org.apache.skywalking.apm.collector.core.module.Module;
 import org.apache.skywalking.apm.collector.core.module.ModuleProvider;
 import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
 import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.table.register.Application;
+import org.apache.skywalking.apm.collector.storage.table.register.Instance;
+import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress;
+import org.apache.skywalking.apm.collector.storage.table.register.ServiceName;
 
 /**
  * @author peng-yongsheng
@@ -68,6 +73,8 @@ public class AnalysisRegisterModuleProvider extends ModuleProvider {
 
         graphCreate(workerCreateListener);
 
+        registerRemoteData();
+
         PersistenceTimer persistenceTimer = new PersistenceTimer(AnalysisRegisterModule.NAME);
         persistenceTimer.start(getManager(), workerCreateListener.getPersistenceWorkers());
     }
@@ -93,4 +100,12 @@ public class AnalysisRegisterModuleProvider extends ModuleProvider {
         NetworkAddressRegisterGraph networkAddressRegisterGraph = new NetworkAddressRegisterGraph(getManager(), workerCreateListener);
         networkAddressRegisterGraph.create();
     }
+
+    private void registerRemoteData() {
+        RemoteDataRegisterService remoteDataRegisterService = getManager().find(RemoteModule.NAME).getService(RemoteDataRegisterService.class);
+        remoteDataRegisterService.register(Application.class, new Application.InstanceCreator());
+        remoteDataRegisterService.register(Instance.class, new Instance.InstanceCreator());
+        remoteDataRegisterService.register(NetworkAddress.class, new NetworkAddress.InstanceCreator());
+        remoteDataRegisterService.register(ServiceName.class, new ServiceName.InstanceCreator());
+    }
 }

apm-collector/apm-collector-remote/collector-remote-define/src/main/java/org/apache/skywalking/apm/collector/remote/service/RemoteDataRegisterService.java

@@ -28,6 +28,6 @@ public interface RemoteDataRegisterService extends Service {
     void register(Class<? extends RemoteData> dataClass, RemoteDataInstanceCreator instanceCreator);
 
     interface RemoteDataInstanceCreator<REMOTE_DATA extends RemoteData> {
-        REMOTE_DATA createInstance(String id);
+        REMOTE_DATA createInstance();
     }
 }

apm-collector/apm-collector-remote/collector-remote-grpc-provider/src/main/java/org/apache/skywalking/apm/collector/remote/grpc/handler/RemoteCommonServiceHandler.java

@@ -21,7 +21,6 @@ package org.apache.skywalking.apm.collector.remote.grpc.handler;
 import io.grpc.stub.StreamObserver;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.graph.Next;
-import org.apache.skywalking.apm.collector.core.util.Const;
 import org.apache.skywalking.apm.collector.remote.grpc.proto.Empty;
 import org.apache.skywalking.apm.collector.remote.grpc.proto.RemoteCommonServiceGrpc;
 import org.apache.skywalking.apm.collector.remote.grpc.proto.RemoteData;
@@ -58,7 +57,7 @@ public class RemoteCommonServiceHandler extends RemoteCommonServiceGrpc.RemoteCo
                 RemoteData remoteData = message.getRemoteData();
 
                 try {
-                    org.apache.skywalking.apm.collector.core.data.RemoteData output = instanceCreatorGetter.getInstanceCreator(remoteDataId).createInstance(Const.EMPTY_STRING);
+                    org.apache.skywalking.apm.collector.core.data.RemoteData output = instanceCreatorGetter.getInstanceCreator(remoteDataId).createInstance();
                     service.deserialize(remoteData, output);
                     Next next = GraphManager.INSTANCE.findGraph(graphId).toFinder().findNext(nodeId);
                     next.execute(output);

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/alarm/ApplicationAlarm.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.alarm;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -114,4 +116,10 @@ public class ApplicationAlarm extends StreamData implements Alarm {
     public void setAlarmContent(String alarmContent) {
         setDataString(1, alarmContent);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ApplicationAlarm();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/alarm/ApplicationAlarmList.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.alarm;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -107,4 +109,10 @@ public class ApplicationAlarmList extends StreamData {
     public void setTimeBucket(Long timeBucket) {
         setDataLong(0, timeBucket);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ApplicationAlarmList();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/alarm/InstanceAlarm.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.alarm;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -123,4 +125,10 @@ public class InstanceAlarm extends StreamData implements Alarm {
     public void setAlarmContent(String alarmContent) {
         setDataString(1, alarmContent);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new InstanceAlarm();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/alarm/InstanceAlarmList.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.alarm;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -115,4 +117,10 @@ public class InstanceAlarmList extends StreamData {
     public void setAlarmContent(String alarmContent) {
         setDataString(1, alarmContent);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new InstanceAlarmList();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/alarm/ServiceAlarm.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.alarm;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -132,4 +134,10 @@ public class ServiceAlarm extends StreamData implements Alarm {
     public void setAlarmContent(String alarmContent) {
         setDataString(1, alarmContent);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ServiceAlarm();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/alarm/ServiceAlarmList.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.alarm;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -124,4 +126,10 @@ public class ServiceAlarmList extends StreamData {
     public void setAlarmContent(String alarmContent) {
         setDataString(1, alarmContent);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ServiceAlarmList();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationComponent.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.application;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -89,4 +91,10 @@ public class ApplicationComponent extends StreamData {
     public void setApplicationId(Integer applicationId) {
         setDataInteger(1, applicationId);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ApplicationComponent();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationMapping.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.application;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -89,4 +91,10 @@ public class ApplicationMapping extends StreamData {
     public void setTimeBucket(long timeBucket) {
         setDataLong(0, timeBucket);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ApplicationMapping();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationMetric.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.application;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.data.operator.AddOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
 import org.apache.skywalking.apm.collector.storage.table.Metric;
 
 /**
@@ -254,4 +256,10 @@ public class ApplicationMetric extends StreamData implements Metric {
     public void setFrustratedCount(long frustratedCount) {
         setDataLong(15, frustratedCount);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ApplicationMetric();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationReferenceMetric.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.application;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.data.operator.AddOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
 import org.apache.skywalking.apm.collector.storage.table.Metric;
 
 /**
@@ -263,4 +265,10 @@ public class ApplicationReferenceMetric extends StreamData implements Metric {
     public void setFrustratedCount(long frustratedCount) {
         setDataLong(15, frustratedCount);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ApplicationReferenceMetric();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/instance/InstanceMapping.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.instance;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -98,4 +100,10 @@ public class InstanceMapping extends StreamData {
     public void setTimeBucket(long timeBucket) {
         setDataLong(0, timeBucket);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new InstanceMapping();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/instance/InstanceMetric.java

@@ -19,10 +19,12 @@
 package org.apache.skywalking.apm.collector.storage.table.instance;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.data.operator.AddOperation;
 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.remote.service.RemoteDataRegisterService;
 import org.apache.skywalking.apm.collector.storage.table.Metric;
 
 /**
@@ -237,4 +239,10 @@ public class InstanceMetric extends StreamData implements Metric {
     public void setMqTransactionErrorDurationSum(Long mqTransactionErrorDurationSum) {
         setDataLong(12, mqTransactionErrorDurationSum);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new InstanceMetric();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/instance/InstanceReferenceMetric.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.instance;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.data.operator.AddOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
 import org.apache.skywalking.apm.collector.storage.table.Metric;
 
 /**
@@ -254,4 +256,10 @@ public class InstanceReferenceMetric extends StreamData implements Metric {
     public void setMqTransactionErrorDurationSum(Long mqTransactionErrorDurationSum) {
         setDataLong(12, mqTransactionErrorDurationSum);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new InstanceReferenceMetric();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/Application.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.register;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -105,4 +107,10 @@ public class Application extends StreamData {
     public void setIsAddress(int isAddress) {
         setDataInteger(3, isAddress);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new Application();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/Instance.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.register;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -142,4 +144,10 @@ public class Instance extends StreamData {
     public void setIsAddress(int isAddress) {
         setDataInteger(3, isAddress);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new Instance();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddress.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.register;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -97,4 +99,10 @@ public class NetworkAddress extends StreamData {
     public void setServerType(Integer serverType) {
         setDataInteger(2, serverType);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new NetworkAddress();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/ServiceName.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.register;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 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.remote.service.RemoteDataRegisterService;
 
 /**
  * @author peng-yongsheng
@@ -96,4 +98,10 @@ public class ServiceName extends StreamData {
     public void setSrcSpanType(int srcSpanType) {
         setDataInteger(2, srcSpanType);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ServiceName();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/service/ServiceMetric.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.service;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.data.operator.AddOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
 import org.apache.skywalking.apm.collector.storage.table.Metric;
 
 /**
@@ -245,4 +247,10 @@ public class ServiceMetric extends StreamData implements Metric {
     public void setMqTransactionErrorDurationSum(Long mqTransactionErrorDurationSum) {
         setDataLong(12, mqTransactionErrorDurationSum);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ServiceMetric();
+        }
+    }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/service/ServiceReferenceMetric.java

@@ -19,9 +19,11 @@
 package org.apache.skywalking.apm.collector.storage.table.service;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
+import org.apache.skywalking.apm.collector.core.data.RemoteData;
 import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.data.operator.AddOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
+import org.apache.skywalking.apm.collector.remote.service.RemoteDataRegisterService;
 import org.apache.skywalking.apm.collector.storage.table.Metric;
 
 /**
@@ -275,4 +277,10 @@ public class ServiceReferenceMetric extends StreamData implements Metric {
     public void setMqTransactionErrorDurationSum(Long mqTransactionErrorDurationSum) {
         setDataLong(12, mqTransactionErrorDurationSum);
     }
+
+    public static class InstanceCreator implements RemoteDataRegisterService.RemoteDataInstanceCreator {
+        @Override public RemoteData createInstance() {
+            return new ServiceReferenceMetric();
+        }
+    }
 }