From 8051965b8675a067aa7897f4673232a5d682a73b Mon Sep 17 00:00:00 2001 From: laynexiong Date: Thu, 17 Sep 2026 11:17:17 +0800 Subject: [PATCH 1/2] feat: use a map with a capacity limit to prevent OutOfMemoryError (OOM). --- trpc-proto/trpc-proto-standard/pom.xml | 4 + .../standard/common/StandardServerCodec.java | 125 +++++++- .../common/StandardServerCodecCacheTest.java | 297 ++++++++++++++++++ 3 files changed, 409 insertions(+), 17 deletions(-) create mode 100644 trpc-proto/trpc-proto-standard/src/test/java/com/tencent/trpc/proto/standard/common/StandardServerCodecCacheTest.java diff --git a/trpc-proto/trpc-proto-standard/pom.xml b/trpc-proto/trpc-proto-standard/pom.xml index 0d5b411e9..de1904761 100644 --- a/trpc-proto/trpc-proto-standard/pom.xml +++ b/trpc-proto/trpc-proto-standard/pom.xml @@ -42,6 +42,10 @@ + + com.github.ben-manes.caffeine + caffeine + commons-codec commons-codec diff --git a/trpc-proto/trpc-proto-standard/src/main/java/com/tencent/trpc/proto/standard/common/StandardServerCodec.java b/trpc-proto/trpc-proto-standard/src/main/java/com/tencent/trpc/proto/standard/common/StandardServerCodec.java index 68d86b19e..f80e65d00 100644 --- a/trpc-proto/trpc-proto-standard/src/main/java/com/tencent/trpc/proto/standard/common/StandardServerCodec.java +++ b/trpc-proto/trpc-proto-standard/src/main/java/com/tencent/trpc/proto/standard/common/StandardServerCodec.java @@ -13,6 +13,8 @@ import static com.tencent.trpc.core.rpc.RpcContextValueKeys.SERVER_SIGNATURE_VERIFY_RESULT_KEY; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import com.google.protobuf.ByteString; import com.tencent.trpc.core.common.config.ProtocolConfig; import com.tencent.trpc.core.compressor.spi.Compressor; @@ -41,9 +43,9 @@ import com.tencent.trpc.proto.standard.common.TRPCProtocol.ResponseProtocol.Builder; import com.tencent.trpc.proto.standard.common.TRPCProtocol.TrpcCallType; import com.tencent.trpc.proto.standard.common.TRPCProtocol.TrpcMessageType; -import java.util.Map; +import java.time.Duration; import java.util.Objects; -import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Function; import org.apache.commons.lang3.StringUtils; /** @@ -52,19 +54,100 @@ public class StandardServerCodec extends ServerCodec { /** - * Remotely call the cache of services and methods to avoid string cutting operations during decoding + * The max entry number of the decoding caches, the entries will be evicted when it is exceeded. */ - private static final Map FUNC_INFO_CACHE = new ConcurrentHashMap<>(); + private static final int CACHE_MAX_SIZE = 10000; + + /** + * The initial capacity of the decoding caches. + */ + private static final int CACHE_INITIAL_CAPACITY = 64; + + /** + * The expire time(in minutes) after the last access of a decoding cache entry. + */ + private static final int CACHE_EXPIRE_MINUTES = 60; + + /** + * The max length of the cache key, the key longer than it will never be cached, so that a few oversized + * attacker-controlled keys can not occupy too much memory. + */ + private static final int CACHE_KEY_MAX_LENGTH = 1024; + + /** + * The separator of the func of the request head, whose format is {@code /serviceName/methodName}. + */ + private static final String FUNC_SEPARATOR = "/"; + + /** + * The begin index of the serviceName in the func, the func always starts with {@link #FUNC_SEPARATOR}. + */ + private static final int FUNC_SERVICE_NAME_BEGIN_INDEX = 1; + + /** + * The index of the serviceName in the parsed func info. + */ + private static final int FUNC_SERVICE_NAME_INDEX = 0; + + /** + * The index of the methodName in the parsed func info. + */ + private static final int FUNC_METHOD_NAME_INDEX = 1; + + /** + * Remotely call the cache of services and methods to avoid string cutting operations during decoding. + * + *

Note: the cache key comes from the attacker-controllable protocol head, so it MUST be bounded, see + * {@link #newCache()} and {@link #getOrCompute(Cache, String, Function)}.

+ */ + private static final Cache FUNC_INFO_CACHE = newCache(); /** * The cache of the caller and callee information avoids the string cutting operation during decoding, which can * increase the throughput of the framework by about 4%. Considering that the mainstream of the current - * architecture is microservices, and each service has limited external interfaces, the built-in ConcurrentHashMap - * is used as a cache here. If the number of caches is too large, you can consider migrating to caffeine, - * but this will greatly offset the performance optimization here. + * architecture is microservices, and each service has limited external interfaces, a bounded cache is enough + * here. + * + *

Note: the cache key comes from the attacker-controllable protocol head, so it MUST be bounded, see + * {@link #newCache()} and {@link #getOrCompute(Cache, String, Function)}.

*/ - private static final Map CALL_INFO_CACHE = new ConcurrentHashMap<>(); + private static final Cache CALL_INFO_CACHE = newCache(); + /** + * Create a bounded cache whose entries are evicted by size and by idle time. + * + * @param the type of the cached value + * @return the bounded cache + */ + private static Cache newCache() { + return Caffeine.newBuilder() + .initialCapacity(CACHE_INITIAL_CAPACITY) + .maximumSize(CACHE_MAX_SIZE) + .expireAfterAccess(Duration.ofMinutes(CACHE_EXPIRE_MINUTES)) + .build(); + } + + /** + * Get the value from the bounded cache, compute it if absent. + * + *

The cache keys are built from the request head fields(func/caller/callee) which are fully controlled by the + * remote peer, and they are written before the service/method existence check. An unbounded cache would allow an + * attacker to write an entry per request and finally exhaust the heap(OOM). The cache is bounded in two + * dimensions here: the entry number is limited by {@link #CACHE_MAX_SIZE}, and an oversized key is never cached + * so that the memory of a single entry is limited as well.

+ * + * @param cache the cache to read and write + * @param key the cache key, which is untrusted + * @param mappingFunction the function to compute the value + * @param the type of the cached value + * @return the cached or newly computed value + */ + private static V getOrCompute(Cache cache, String key, Function mappingFunction) { + if (key.length() > CACHE_KEY_MAX_LENGTH) { + return mappingFunction.apply(key); + } + return cache.get(key, mappingFunction); + } @Override public void encode(Channel channel, ChannelBuffer channelBuffer, Object message) { @@ -236,21 +319,29 @@ private void setAttachments(RequestProtocol requestHeader, DefRequest request) { private RpcInvocation buildRpcInvocation(StandardPackage packet, RequestProtocol requestHeader) { RpcInvocation inv = new RpcInvocation(); String func = requestHeader.getFunc().toStringUtf8(); - String[] funcInfo = FUNC_INFO_CACHE.computeIfAbsent(func, s -> { - int idx = func.lastIndexOf("/"); - // func format: /serviceName/methodName - return (idx > 1 && func.length() > idx + 1) ? new String[]{func.substring(1, idx), func.substring(idx + 1)} - : new String[]{"", ""}; - }); + String[] funcInfo = getOrCompute(FUNC_INFO_CACHE, func, StandardServerCodec::parseFunc); inv.setFunc(func); - inv.setRpcServiceName(funcInfo[0]); - inv.setRpcMethodName(funcInfo[1]); + inv.setRpcServiceName(funcInfo[FUNC_SERVICE_NAME_INDEX]); + inv.setRpcMethodName(funcInfo[FUNC_METHOD_NAME_INDEX]); Object[] obj = new Object[]{new DecodableValue(requestHeader.getContentEncoding(), requestHeader.getContentType(), packet.getBodyBytes())}; inv.setArguments(obj); return inv; } + /** + * Parse the func of the request head, whose format is {@code /serviceName/methodName}. + * + * @param func the func of the request head + * @return an array of [serviceName, methodName], both of them are empty if the func is illegal + */ + private static String[] parseFunc(String func) { + int idx = func.lastIndexOf(FUNC_SEPARATOR); + return (idx > FUNC_SERVICE_NAME_BEGIN_INDEX && func.length() > idx + 1) + ? new String[]{func.substring(FUNC_SERVICE_NAME_BEGIN_INDEX, idx), func.substring(idx + 1)} + : new String[]{StringUtils.EMPTY, StringUtils.EMPTY}; + } + private void setDyeingKeyIfNonNull(RequestProtocol requestHeader, DefRequest request) { ByteString dyeingKeyByte = requestHeader.getTransInfoMap().get(TrpcTransInfoKeys.DYEING_KEY); if (dyeingKeyByte != null) { @@ -276,7 +367,7 @@ private CallInfo buildCallInfo(RpcInvocation rpcInvocation, TRPCProtocol.Request if (StringUtils.isBlank(cacheKey)) { return null; } - return CALL_INFO_CACHE.computeIfAbsent(cacheKey, s -> { + return getOrCompute(CALL_INFO_CACHE, cacheKey, s -> { CallInfo callInfo = new CallInfo(); fillCallerInfo(caller, callInfo); fillCalleeInfo(callee, rpcMethodName, callInfo); diff --git a/trpc-proto/trpc-proto-standard/src/test/java/com/tencent/trpc/proto/standard/common/StandardServerCodecCacheTest.java b/trpc-proto/trpc-proto-standard/src/test/java/com/tencent/trpc/proto/standard/common/StandardServerCodecCacheTest.java new file mode 100644 index 000000000..8565cc402 --- /dev/null +++ b/trpc-proto/trpc-proto-standard/src/test/java/com/tencent/trpc/proto/standard/common/StandardServerCodecCacheTest.java @@ -0,0 +1,297 @@ +/* + * Tencent is pleased to support the open source community by making tRPC available. + * + * Copyright (C) 2023 Tencent. + * All rights reserved. + * + * If you have downloaded a copy of the tRPC source code from Tencent, + * please note that tRPC source code is licensed under the Apache 2.0 License, + * A copy of the Apache 2.0 License can be found in the LICENSE file. + */ + +package com.tencent.trpc.proto.standard.common; + +import com.github.benmanes.caffeine.cache.Cache; +import com.google.protobuf.ByteString; +import com.tencent.trpc.core.common.config.ProtocolConfig; +import com.tencent.trpc.core.rpc.CallInfo; +import com.tencent.trpc.core.rpc.Request; +import com.tencent.trpc.proto.standard.common.TRPCProtocol.RequestProtocol; +import com.tencent.trpc.transport.netty.NettyChannel; +import com.tencent.trpc.transport.netty.NettyChannelBuffer; +import io.netty.buffer.UnpooledByteBufAllocator; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.function.Function; +import org.apache.commons.lang3.StringUtils; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +/** + * Test the bounded decoding caches of {@link StandardServerCodec}, which protect the server from the memory + * exhaustion attack: the cache keys(func/caller/callee) are fully controlled by the remote peer and are written + * before the service/method existence check. + */ +public class StandardServerCodecCacheTest { + + private static final String FUNC_INFO_CACHE_FIELD = "FUNC_INFO_CACHE"; + + private static final String CALL_INFO_CACHE_FIELD = "CALL_INFO_CACHE"; + + private static final String CACHE_MAX_SIZE_FIELD = "CACHE_MAX_SIZE"; + + private static final String CACHE_KEY_MAX_LENGTH_FIELD = "CACHE_KEY_MAX_LENGTH"; + + private static final String GET_OR_COMPUTE_METHOD = "getOrCompute"; + + private static final String PARSE_FUNC_METHOD = "parseFunc"; + + private static final String SERVICE_NAME = "helloservice"; + + private static final String METHOD_NAME = "sayHello"; + + private static final String FUNC = "/" + SERVICE_NAME + "/" + METHOD_NAME; + + private static final String CALLER = "trpc.callerApp.callerServer.callerService"; + + private static final String CALLEE = "trpc.calleeApp.calleeServer.calleeService.calleeMethod"; + + private static final int BUFFER_SIZE = 65535; + + private static final int LOCAL_PORT = 125; + + private static final String LOCAL_IP = "127.0.0.1"; + + private Cache funcInfoCache; + + private Cache callInfoCache; + + private int cacheMaxSize; + + private int cacheKeyMaxLength; + + @Before + public void before() throws Exception { + funcInfoCache = getStaticField(FUNC_INFO_CACHE_FIELD); + callInfoCache = getStaticField(CALL_INFO_CACHE_FIELD); + cacheMaxSize = getStaticField(CACHE_MAX_SIZE_FIELD); + cacheKeyMaxLength = getStaticField(CACHE_KEY_MAX_LENGTH_FIELD); + funcInfoCache.invalidateAll(); + callInfoCache.invalidateAll(); + funcInfoCache.cleanUp(); + callInfoCache.cleanUp(); + } + + @Test + public void testParseFuncWithLegalFunc() throws Exception { + String[] funcInfo = parseFunc(FUNC); + Assert.assertEquals(SERVICE_NAME, funcInfo[0]); + Assert.assertEquals(METHOD_NAME, funcInfo[1]); + // only the last separator is used to split the service and the method + String[] multiLevel = parseFunc("/a/b/c"); + Assert.assertEquals("a/b", multiLevel[0]); + Assert.assertEquals("c", multiLevel[1]); + } + + @Test + public void testParseFuncWithIllegalFunc() throws Exception { + String[] illegalFuncs = new String[]{StringUtils.EMPTY, "/", "//", "/abc", "abc", "/abc/"}; + for (String illegalFunc : illegalFuncs) { + String[] funcInfo = parseFunc(illegalFunc); + Assert.assertEquals(illegalFunc, StringUtils.EMPTY, funcInfo[0]); + Assert.assertEquals(illegalFunc, StringUtils.EMPTY, funcInfo[1]); + } + } + + @Test + public void testGetOrComputeReusesCachedValue() throws Exception { + String[] first = getOrCompute(funcInfoCache, FUNC); + String[] second = getOrCompute(funcInfoCache, FUNC); + Assert.assertSame(first, second); + Assert.assertEquals(1, funcInfoCache.estimatedSize()); + Assert.assertNotNull(funcInfoCache.getIfPresent(FUNC)); + } + + @Test + public void testGetOrComputeCachesKeyOfMaxLength() throws Exception { + String key = buildFunc(cacheKeyMaxLength); + Assert.assertEquals(cacheKeyMaxLength, key.length()); + String[] value = getOrCompute(funcInfoCache, key); + Assert.assertEquals(METHOD_NAME, value[1]); + Assert.assertEquals(1, funcInfoCache.estimatedSize()); + Assert.assertSame(value, funcInfoCache.getIfPresent(key)); + } + + @Test + public void testGetOrComputeNeverCachesOversizedKey() throws Exception { + String key = buildFunc(cacheKeyMaxLength + 1); + Assert.assertEquals(cacheKeyMaxLength + 1, key.length()); + String[] first = getOrCompute(funcInfoCache, key); + String[] second = getOrCompute(funcInfoCache, key); + // the value is still computed correctly, but it is never cached + Assert.assertEquals(METHOD_NAME, first[1]); + Assert.assertNotSame(first, second); + Assert.assertNull(funcInfoCache.getIfPresent(key)); + Assert.assertEquals(0, funcInfoCache.estimatedSize()); + } + + @Test + public void testCacheIsBoundedBySize() throws Exception { + int total = cacheMaxSize * 2; + for (int i = 0; i < total; i++) { + getOrCompute(funcInfoCache, "/" + SERVICE_NAME + i + "/" + METHOD_NAME); + } + funcInfoCache.cleanUp(); + Assert.assertTrue("cache size should be bounded, but was " + funcInfoCache.estimatedSize(), + funcInfoCache.estimatedSize() <= cacheMaxSize); + } + + @Test + public void testHotKeyIsKeptWhileCacheIsFlooded() throws Exception { + String[] hot = getOrCompute(funcInfoCache, FUNC); + // the hot key is accessed much more frequently than every flooding key + for (int i = 0; i < cacheMaxSize * 2; i++) { + getOrCompute(funcInfoCache, FUNC); + getOrCompute(funcInfoCache, "/" + SERVICE_NAME + i + "/" + METHOD_NAME); + } + funcInfoCache.cleanUp(); + Assert.assertTrue(funcInfoCache.estimatedSize() <= cacheMaxSize); + Assert.assertSame(hot, funcInfoCache.getIfPresent(FUNC)); + } + + @Test + public void testDecodeParsesAndCachesFuncInfo() { + Request request = decode(buildRequestHead(FUNC, CALLER, CALLEE)); + Assert.assertEquals(FUNC, request.getInvocation().getFunc()); + Assert.assertEquals(SERVICE_NAME, request.getInvocation().getRpcServiceName()); + Assert.assertEquals(METHOD_NAME, request.getInvocation().getRpcMethodName()); + Assert.assertNotNull(funcInfoCache.getIfPresent(FUNC)); + Assert.assertEquals(1, funcInfoCache.estimatedSize()); + // the second decoding of the same func reuses the cached entry + Request another = decode(buildRequestHead(FUNC, CALLER, CALLEE)); + Assert.assertEquals(SERVICE_NAME, another.getInvocation().getRpcServiceName()); + Assert.assertEquals(1, funcInfoCache.estimatedSize()); + } + + @Test + public void testDecodeParsesAndCachesCallInfo() { + Request request = decode(buildRequestHead(FUNC, CALLER, CALLEE)); + CallInfo callInfo = request.getMeta().getCallInfo(); + Assert.assertEquals(CALLER, callInfo.getCaller()); + Assert.assertEquals("callerApp", callInfo.getCallerApp()); + Assert.assertEquals("callerServer", callInfo.getCallerServer()); + Assert.assertEquals("callerService", callInfo.getCallerService()); + Assert.assertEquals(CALLEE, callInfo.getCallee()); + Assert.assertEquals("calleeApp", callInfo.getCalleeApp()); + Assert.assertEquals("calleeServer", callInfo.getCalleeServer()); + Assert.assertEquals("calleeService", callInfo.getCalleeService()); + Assert.assertEquals("calleeMethod", callInfo.getCalleeMethod()); + Assert.assertEquals(1, callInfoCache.estimatedSize()); + // the same caller/callee/method reuses the cached entry + Request another = decode(buildRequestHead(FUNC, CALLER, CALLEE)); + Assert.assertSame(callInfo, another.getMeta().getCallInfo()); + Assert.assertEquals(1, callInfoCache.estimatedSize()); + } + + @Test + public void testDecodeWithOversizedFuncDoesNotPolluteCache() { + String func = buildFunc(cacheKeyMaxLength + 1); + Request request = decode(buildRequestHead(func, CALLER, CALLEE)); + Assert.assertEquals(METHOD_NAME, request.getInvocation().getRpcMethodName()); + Assert.assertNull(funcInfoCache.getIfPresent(func)); + Assert.assertEquals(0, funcInfoCache.estimatedSize()); + } + + @Test + public void testDecodeWithOversizedCallerDoesNotPolluteCache() { + String caller = CALLER + StringUtils.repeat('x', cacheKeyMaxLength); + Request request = decode(buildRequestHead(FUNC, caller, CALLEE)); + Assert.assertEquals(caller, request.getMeta().getCallInfo().getCaller()); + Assert.assertEquals(0, callInfoCache.estimatedSize()); + } + + @Test + public void testDecodeWithBlankCallInfoIsNotCached() { + Request request = decode(buildRequestHead(StringUtils.EMPTY, StringUtils.EMPTY, StringUtils.EMPTY)); + Assert.assertEquals(StringUtils.EMPTY, request.getInvocation().getRpcServiceName()); + Assert.assertEquals(StringUtils.EMPTY, request.getInvocation().getRpcMethodName()); + Assert.assertEquals(0, callInfoCache.estimatedSize()); + Assert.assertNull(callInfoCache.getIfPresent(StringUtils.EMPTY)); + } + + @Test + public void testDecodeWithDistinctFuncKeepsCachesBounded() { + int total = 100; + for (int i = 0; i < total; i++) { + decode(buildRequestHead("/" + SERVICE_NAME + i + "/" + METHOD_NAME, CALLER + i, CALLEE)); + } + funcInfoCache.cleanUp(); + callInfoCache.cleanUp(); + Assert.assertTrue(funcInfoCache.estimatedSize() <= Math.min(total, cacheMaxSize)); + Assert.assertTrue(callInfoCache.estimatedSize() <= Math.min(total, cacheMaxSize)); + } + + private Request decode(RequestProtocol requestHead) { + byte[] headBytes = requestHead.toByteArray(); + StandardPackage pkg = new StandardPackage(); + pkg.setHeadBytes(headBytes); + pkg.getFrame().setHeadSize(headBytes.length); + pkg.getFrame().setSize(StandardFrame.FRAME_SIZE + headBytes.length); + ProtocolConfig config = new ProtocolConfig(); + config.setIp(LOCAL_IP); + config.setPort(LOCAL_PORT); + config.setDefault(); + NettyChannel channel = new NettyChannel(null, config); + NettyChannelBuffer buffer = new NettyChannelBuffer(UnpooledByteBufAllocator.DEFAULT.buffer(BUFFER_SIZE)); + pkg.write(buffer); + return (Request) new StandardServerCodec().decode(channel, buffer); + } + + private RequestProtocol buildRequestHead(String func, String caller, String callee) { + return RequestProtocol.newBuilder() + .setFunc(ByteString.copyFromUtf8(func)) + .setCaller(ByteString.copyFromUtf8(caller)) + .setCallee(ByteString.copyFromUtf8(callee)) + .build(); + } + + /** + * Build a legal func whose total length is the given length. + * + * @param length the expected length of the func + * @return the func like {@code /xxx.../sayHello} + */ + private String buildFunc(int length) { + int padding = length - METHOD_NAME.length() - 2; + return "/" + StringUtils.repeat('x', padding) + "/" + METHOD_NAME; + } + + @SuppressWarnings("unchecked") + private static T getStaticField(String name) throws Exception { + Field field = StandardServerCodec.class.getDeclaredField(name); + field.setAccessible(true); + return (T) field.get(null); + } + + @SuppressWarnings("unchecked") + private static String[] getOrCompute(Cache cache, String key) throws Exception { + Method method = StandardServerCodec.class.getDeclaredMethod(GET_OR_COMPUTE_METHOD, Cache.class, + String.class, Function.class); + method.setAccessible(true); + Function mappingFunction = func -> { + try { + return parseFunc(func); + } catch (Exception e) { + throw new IllegalStateException(e); + } + }; + return (String[]) method.invoke(null, cache, key, mappingFunction); + } + + private static String[] parseFunc(String func) throws Exception { + Method method = StandardServerCodec.class.getDeclaredMethod(PARSE_FUNC_METHOD, String.class); + method.setAccessible(true); + return (String[]) method.invoke(null, func); + } +} From b2f7ba27bfa3d413326862a3e29152863f770501 Mon Sep 17 00:00:00 2001 From: laynexiong Date: Thu, 17 Sep 2026 16:40:32 +0800 Subject: [PATCH 2/2] feat: use a map with a capacity limit to prevent OutOfMemoryError (OOM). --- .../tencent/trpc/proto/http/HttpsRpcClientTest.java | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/trpc-proto/trpc-proto-http/src/test/java/com/tencent/trpc/proto/http/HttpsRpcClientTest.java b/trpc-proto/trpc-proto-http/src/test/java/com/tencent/trpc/proto/http/HttpsRpcClientTest.java index 085e9ebcf..a0fe05d9a 100644 --- a/trpc-proto/trpc-proto-http/src/test/java/com/tencent/trpc/proto/http/HttpsRpcClientTest.java +++ b/trpc-proto/trpc-proto-http/src/test/java/com/tencent/trpc/proto/http/HttpsRpcClientTest.java @@ -26,11 +26,13 @@ import com.tencent.trpc.core.logger.LoggerFactory; import com.tencent.trpc.core.rpc.RpcClientContext; import com.tencent.trpc.core.utils.NetUtils; +import com.tencent.trpc.proto.http.client.AbstractConsumerInvoker; import java.io.File; import java.util.HashMap; import java.util.Map; import org.junit.AfterClass; import org.junit.Assert; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; import tests.service.GreeterJsonService; @@ -51,6 +53,17 @@ public class HttpsRpcClientTest { private static Map extMap = new HashMap<>(); + /** + * The timeout manager of {@link AbstractConsumerInvoker} is a static one shared by the whole JVM, and its + * underlying timer can never be restarted once it is stopped. Other test classes running before this one may + * have stopped it(directly or by the shutdown listener of the container), so it is rebuilt before every test + * case to keep the test cases independent. + */ + @Before + public void beforeTest() { + AbstractConsumerInvoker.reset(); + } + @BeforeClass public static void startHttpServer() {