Fix regression bug caused by sharing TSerializer across multiple threads
@@ -56,7 +56,9 @@ public class ServerChannels {
public static final String CHANNEL_LOCK_TIMEOUT_MSG = "Timeout while acquiring channel lock";
private final QueryRouter _queryRouter;
private final BrokerMetrics _brokerMetrics;
- private final TSerializer _serializer = new TSerializer(new TCompactProtocol.Factory());
+ // TSerializer currently is not thread safe, must be put into a ThreadLocal.
+ private final ThreadLocal<TSerializer> _threadLocalTSerializer =
+ ThreadLocal.withInitial(() -> new TSerializer(new TCompactProtocol.Factory()));
private final ConcurrentHashMap<ServerRoutingInstance, ServerChannel> _serverToChannelMap = new ConcurrentHashMap<>();
private final EventLoopGroup _eventLoopGroup = new NioEventLoopGroup();
private final TlsConfig _tlsConfig;
@@ -87,7 +89,7 @@ public class ServerChannels {
public void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryResponse,
ServerRoutingInstance serverRoutingInstance, InstanceRequest instanceRequest, long timeoutMs)
throws Exception {
- byte[] requestBytes = _serializer.serialize(instanceRequest);
+ byte[] requestBytes = _threadLocalTSerializer.get().serialize(instanceRequest);
_serverToChannelMap.computeIfAbsent(serverRoutingInstance, ServerChannel::new)
.sendRequest(rawTableName, asyncQueryResponse, serverRoutingInstance, requestBytes, timeoutMs);
}
