MicroserviceVersions存在并发问题

[yhs0092]

测试版本

2.8.2

问题原理分析

问题出在 org.apache.servicecomb.registry.consumer.MicroserviceVersions#setInstances 方法中:

      versions.forEach((key, value) -> value.setInstances(new ArrayList<>()));  // 步骤1, 将 MicroserviceVersion 内的实例列表置空
      for (Entry<String, List<MicroserviceInstance>> entry : mergedInstances.microserviceIdMap.entrySet()) {
        // always update microservice versions, because we allow microservice info override, like schema info
        MicroserviceVersion newVersion = createMicroserviceVersion(entry.getKey(), entry.getValue());
        newVersion.setInstances(entry.getValue());
        versions.put(entry.getKey(), newVersion);  // 步骤2, 用新 MicroserviceVersion 替换旧的空 MicroserviceVersion
      }

如上述代码片段, 步骤1和步骤2之间, 其他执行微服务调用逻辑的线程从 MicroserviceVersions 中拿到的 MicroserviceVersion 是空的, 这会影响org.apache.servicecomb.core.registry.discovery.OperationInstancesDiscoveryFilter#sortedMicroserviceVersion方法的逻辑, 导致OperationInstancesDiscoveryFilter返回的可用实例列表为空, 业务微服务调用报错.

[yhs0092]

测试用例

package org.apache.servicecomb.registry.consumer;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.servicecomb.registry.api.registry.Microservice;
import org.apache.servicecomb.registry.api.registry.MicroserviceInstance;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

class MicroserviceVersionsTest {

  private static final int SCALE = 10000;

  private static final String MOCK_SERVICE_ID = "serviceId";

  private static final String MOCK_INSTANCE_ID = "instanceId";

  @Test
  public void test() throws InterruptedException {
    final MicroserviceVersions microserviceVersions = mockMicroserviceVersions();
    final List<MicroserviceInstance> instanceList = mockMicroserviceInstanceList();
    final AtomicLong counter = new AtomicLong(0);
    microserviceVersions.safeSetInstances(instanceList, "" + counter.getAndIncrement());

    final ExecutorService pool = Executors.newFixedThreadPool(2);
    final CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
    final CountDownLatch countDownLatch = new CountDownLatch(2);
    final AtomicBoolean continueFlag = new AtomicBoolean(true);
    pool.submit(() -> {
      // mock pulled instance list refreshing
      try {
        cyclicBarrier.await();
      } catch (InterruptedException | BrokenBarrierException e) {
        e.printStackTrace();
      }

      while (continueFlag.get()) {
        microserviceVersions.safeSetInstances(instanceList, "" + counter.getAndIncrement());
        sleep();
      }
      countDownLatch.countDown();
    });
    final HashMap<Integer, Integer> resultMap = new HashMap<>();
    pool.submit(() -> {
      // mock microservice invocation load balancing
      try {
        cyclicBarrier.await();
      } catch (InterruptedException | BrokenBarrierException e) {
        e.printStackTrace();
      }

      for (int i = 0; i < SCALE; ++i) {
        sleep();
        resultMap.compute(microserviceVersions.getVersions().get(MOCK_SERVICE_ID).getInstances().size(),
            (k, v) -> v == null ? 1 : v + 1);
      }
      continueFlag.set(false);
      countDownLatch.countDown();
    });
    countDownLatch.await();
    System.out.println(resultMap);
    pool.shutdown();
    Assertions.assertTrue((int) resultMap.get(0) == 0, "concurrent problem occurs!");
  }

  private void sleep() {
    try {
      Thread.sleep(1);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  private MicroserviceVersions mockMicroserviceVersions() {
    return new MicroserviceVersions(new AppManager(), "appId", "ms") {
      @Override
      public MicroserviceVersion createMicroserviceVersion(String microserviceId,
          List<MicroserviceInstance> instances) {
        final Microservice microservice = new Microservice();
        microservice.setVersion("0.0.1.0");
        return new MicroserviceVersion(this, microservice, "msName", instances);
      }
    };
  }

  private List<MicroserviceInstance> mockMicroserviceInstanceList() {
    final MicroserviceInstance instance = new MicroserviceInstance();
    instance.setServiceId(MOCK_SERVICE_ID);
    instance.setInstanceId(MOCK_INSTANCE_ID);
    return Collections.singletonList(instance);
  }
}

测试结果

[vegetableYang]

我们服务在起多个实例并发调用时会偶现请求失败，怀疑和这个问题是同个问题，客户端报错：
[11:45:05:106] [DEBUG] [] - org.springframework.core.log.CompositeLog.debug(CompositeLog.java:147) - HTTP POST cse://xxx-service/xxx/v1/task/update/202302091144 [11:45:05:107] [DEBUG] [] - org.springframework.core.log.CompositeLog.debug(CompositeLog.java:147) - Accept=[*/*] [11:45:05:107] [DEBUG] [] - org.springframework.core.log.CompositeLog.debug(CompositeLog.java:147) - Writing [1] as "application/json" [11:45:05:108] [INFO] [] - org.apache.servicecomb.registry.consumer.MicroserviceVersions.<init>(MicroserviceVersions.java:97) - create MicroserviceVersions, appId=APPN, microserviceName=xxx-service. [11:45:05:111] [WARN] [] - org.apache.servicecomb.serviceregistry.client.http.ServiceRegistryClientImpl.lambda$null$7(ServiceRegistryClientImpl.java:277) - failed to findInstances: {"errorCode":"400012","errorMessage":"Micro-service does not exist","detail":"Consumer[21c******] does not exist."} [11:45:05:112] [INFO] [] - org.apache.servicecomb.serviceregistry.registry.cache.RefreshableServiceRegistryCache.removeCacheIfServiceNotFound(RefreshableServiceRegistryCache.java:130) - microserviceCache[xxx-service@APPN@@0.0.0.0+] got removed [11:45:05:112] [INFO] [] - org.apache.servicecomb.serviceregistry.registry.cache.AggregateServiceRegistryCache.removeMicroserviceCacheIfNotExist(AggregateServiceRegistryCache.java:91) - microserviceCache[xxx-service@APPN@@0.0.0.0+] got removed [11:45:05:113] [INFO] [] - org.apache.servicecomb.registry.consumer.MicroserviceManager.tryRemoveInvalidMicroservice(MicroserviceManager.java:105) - remove microservice, appId=APPN, microserviceName=xxx-service. [11:45:05:114] [DEBUG] [] - org.apache.servicecomb.config.priority.PriorityProperty.updateFinalValue(PriorityProperty.java:171) - config inited, "[servicecomb.references.version-rule.xxx-service, servicecomb.references.xxx-service.version-rule, servicecomb.references.version-rule]" set to 0+, effective key is "servicecomb.references.version-rule". [11:45:05:114] [INFO] [] - org.apache.servicecomb.registry.consumer.MicroserviceVersions.createAndInitMicroserviceVersionRule(MicroserviceVersions.java:289) - create MicroserviceVersionRule, appId=APPN, microserviceName=xxx-service, versionRule=0.0.0.0+. [11:45:05:115] [INFO] [] - org.apache.servicecomb.registry.consumer.MicroserviceVersionRule.printData(MicroserviceVersionRule.java:113) - update instances to be empty caused by version rule, appId=APPN, microserviceName=xxx-service, versionRule=0.0.0.0+, latestVersion=null, inputVersionCount=0, inputInstanceCount=0 [11:45:05:117] [ERROR] [] - org.apache.servicecomb.core.tracing.TraceIdLogger.error(TraceIdLogger.java:50) - unexpected error operation=PRODUCER rest client-service.start.startTask, message=null, cause=Probably invoke a service before it is registered, or no instance found for it, appId=APPN, name=xxx-service, versionRule=0.0.0.0+. [11:45:05:119] [DEBUG] [] - org.jboss.logging.LoggerProviders.logProvider(LoggerProviders.java:152) - Logging Provider: org.jboss.logging.Log4j2LoggerProvider [11:45:05:120] [INFO] [] - org.hibernate.validator.internal.util.Version.<clinit>(Version.java:21) - HV000001: Hibernate Validator 6.2.0.Final [11:45:05:127] [INFO] [] - org.apache.servicecomb.foundation.common.utils.SPIServiceUtils.loadSortedService(SPIServiceUtils.java:79) - Found SPI service org.apache.servicecomb.swagger.invocation.exception.ExceptionToProducerResponseConverter, count=7. [11:45:05:127] [INFO] [] - org.apache.servicecomb.foundation.common.utils.SPIServiceUtils.loadSortedService(SPIServiceUtils.java:82) - 0. org.apache.servicecomb.swagger.invocation.validator.ConstraintViolationExceptionToProducerResponseConverter. [11:45:05:127] [INFO] [] - org.apache.servicecomb.foundation.common.utils.SPIServiceUtils.loadSortedService(SPIServiceUtils.java:82) - 1. com.AAA.APPN.sdk.context.response.cse.CheckFailedExceptionToProducerResponseConverter. [11:45:05:127] [INFO] [] - org.apache.servicecomb.foundation.common.utils.SPIServiceUtils.loadSortedService(SPIServiceUtils.java:82) - 2. com.AAA.APPN.sdk.context.response.cse.DefaultExceptionToProducerResponseConverter. [11:45:05:128] [INFO] [] - org.apache.servicecomb.foundation.common.utils.SPIServiceUtils.loadSortedService(SPIServiceUtils.java:82) - 3. org.apache.servicecomb.swagger.invocation.exception.InvocationExceptionToProducerResponseConverter. [11:45:05:128] [INFO] [] - org.apache.servicecomb.foundation.common.utils.SPIServiceUtils.loadSortedService(SPIServiceUtils.java:82) - 4. org.apache.servicecomb.swagger.invocation.exception.DefaultExceptionToProducerResponseConverter. [11:45:05:128] [INFO] [] - org.apache.servicecomb.foundation.common.utils.SPIServiceUtils.loadSortedService(SPIServiceUtils.java:82) - 5. org.apache.servicecomb.core.exception.IllegalArgumentExceptionToProducerResponseConverter. [11:45:05:128] [INFO] [] - org.apache.servicecomb.foundation.common.utils.SPIServiceUtils.loadSortedService(SPIServiceUtils.java:82) - 6. org.apache.servicecomb.core.exception.ThrowableExceptionToProducerResponseConverter. [11:45:05:130] [ERROR] [] - com.AAA.APPN.sdk.context.response.RestDataBuilder.handleException(RestDataBuilder.java:44) - #handleException; java.lang.IllegalStateException: Probably invoke a service before it is registered, or no instance found for it, appId=APPN, name=xxx-service, versionRule=0.0.0.0+. at org.apache.servicecomb.core.provider.consumer.MicroserviceReferenceConfig.getLatestMicroserviceMeta(MicroserviceReferenceConfig.java:79) ~[java-chassis-core-2.1.5-h0.gts.dsp.r78.jar:2.1.5-h0.gts.dsp.r78] at org.apache.servicecomb.provider.springmvc.reference.CseClientHttpRequest.createRequestMeta(CseClientHttpRequest.java:185) ~[provider-springmvc-2.1.5-h0.gts.dsp.r78.jar:2.1.5-h0.gts.dsp.r78] at org.apache.servicecomb.provider.springmvc.reference.CseClientHttpRequest.execute(CseClientHttpRequest.java:169) ~[provider-springmvc-2.1.5-h0.gts.dsp.r78.jar:2.1.5-h0.gts.dsp.r78] at org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:776) ~[spring-web-5.3.20-h1.jar:5.3.20-h1] at org.springframework.web.client.RestTemplate.execute(RestTemplate.java:711) ~[spring-web-5.3.20-h1.jar:5.3.20-h1] at org.springframework.web.client.RestTemplate.exchange(RestTemplate.java:631) ~[spring-web-5.3.20-h1.jar:5.3.20-h1] at org.apache.servicecomb.provider.springmvc.reference.RestTemplateWrapper.exchange(RestTemplateWrapper.java:196) ~[provider-springmvc-2.1.5-h0.gts.dsp.r78.jar:2.1.5-h0.gts.dsp.r78] at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) ~[spring-core-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:793) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.aspectj.AspectJAfterThrowingAdvice.invoke(AspectJAfterThrowingAdvice.java:64) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:97) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:708) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) ~[spring-core-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:793) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1] at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.20-h1.jar:5.3.20-h1]

[liubao68]

see #3639