#10019

Committed 07 Sep 2023 04:50AM UTC coverage: 47.489% (-0.2%) from 47.655%

Build # #10019

Build Type

push

travis_ci

Committed by

web-flow

Commit Message

Pipe: Fix ConcurrentModificationException caused by concurrently iterating through CachedSchemaPatternMatcher.extractors when an PipeHeartbeatEvent is being assigned (#11074)

* try to fix ConcurrentModificationException when assigning PipeHeartbeatEvent

* Update CachedSchemaPatternMatcher.java

---------

Co-authored-by: 马子坤 <55695098+DanielWang2035@users.noreply.github.com>

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

80551 of 169622 relevant lines covered (47.49%)

0.47 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/MLNodeRPCService.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */

package org.apache.iotdb.db.service;

import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.exception.runtime.RPCServiceException;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.service.ThriftService;
import org.apache.iotdb.commons.service.ThriftServiceThread;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.protocol.thrift.handler.MLNodeRPCServiceThriftHandler;
import org.apache.iotdb.db.protocol.thrift.impl.MLNodeRPCServiceImpl;
import org.apache.iotdb.mpp.rpc.thrift.IMLNodeInternalRPCService;

public class MLNodeRPCService extends ThriftService implements MLNodeRPCServiceMBean {

  private MLNodeRPCServiceImpl impl;

  private MLNodeRPCService() {}

  @Override
  public ServiceType getID() {
    return ServiceType.MLNode_RPC_SERVICE;
  }

  @Override
  public void initTProcessor() {
    impl = new MLNodeRPCServiceImpl();
    initSyncedServiceImpl(null);
    processor = new IMLNodeInternalRPCService.Processor<>(impl);
  }

  @Override
  public void initThriftServiceThread()
      throws IllegalAccessException, InstantiationException, ClassNotFoundException {
    try {
      IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
      thriftServiceThread =
          new ThriftServiceThread(
              processor,
              getID().getName(),
              ThreadName.MLNODE_RPC_SERVICE.getName(),
              getBindIP(),
              getBindPort(),
              config.getRpcMaxConcurrentClientNum(),
              config.getThriftServerAwaitTimeForStopService(),
              new MLNodeRPCServiceThriftHandler(impl),
              false);
    } catch (RPCServiceException e) {
      throw new IllegalAccessException(e.getMessage());
    }
    thriftServiceThread.setName(ThreadName.MLNODE_RPC_SERVICE.getName());
  }

  @Override
  public String getBindIP() {
    return IoTDBDescriptor.getInstance().getConfig().getRpcAddress();
  }

  @Override
  public int getBindPort() {
    return IoTDBDescriptor.getInstance().getConfig().getMLNodePort();
  }

  private static class MLNodeRPCServiceHolder {
    private static final MLNodeRPCService INSTANCE = new MLNodeRPCService();

    private MLNodeRPCServiceHolder() {}
  }

  public static MLNodeRPCService getInstance() {
    return MLNodeRPCService.MLNodeRPCServiceHolder.INSTANCE;
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one
3	* or more contributor license agreements. See the NOTICE file
4	* distributed with this work for additional information
5	* regarding copyright ownership. The ASF licenses this file
6	* to you under the Apache License, Version 2.0 (the
7	* "License"); you may not use this file except in compliance
8	* with the License. You may obtain a copy of the License at
9	*
10	* http://www.apache.org/licenses/LICENSE-2.0
11	*
12	* Unless required by applicable law or agreed to in writing,
13	* software distributed under the License is distributed on an
14	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15	* KIND, either express or implied. See the License for the
16	* specific language governing permissions and limitations
17	* under the License.
18	*/
19
20	package org.apache.iotdb.db.service;
21
22	import org.apache.iotdb.commons.concurrent.ThreadName;
23	import org.apache.iotdb.commons.exception.runtime.RPCServiceException;
24	import org.apache.iotdb.commons.service.ServiceType;
25	import org.apache.iotdb.commons.service.ThriftService;
26	import org.apache.iotdb.commons.service.ThriftServiceThread;
27	import org.apache.iotdb.db.conf.IoTDBConfig;
28	import org.apache.iotdb.db.conf.IoTDBDescriptor;
29	import org.apache.iotdb.db.protocol.thrift.handler.MLNodeRPCServiceThriftHandler;
30	import org.apache.iotdb.db.protocol.thrift.impl.MLNodeRPCServiceImpl;
31	import org.apache.iotdb.mpp.rpc.thrift.IMLNodeInternalRPCService;
32
33	public class MLNodeRPCService extends ThriftService implements MLNodeRPCServiceMBean {
34
35	private MLNodeRPCServiceImpl impl;
36
37	private MLNodeRPCService() {}
38
39	@Override
40	public ServiceType getID() {
41	return ServiceType.MLNode_RPC_SERVICE;	×
42	}
43
44	@Override
45	public void initTProcessor() {
46	impl = new MLNodeRPCServiceImpl();	×
47	initSyncedServiceImpl(null);	×
48	processor = new IMLNodeInternalRPCService.Processor<>(impl);	×
49	}	×
50
51	@Override
52	public void initThriftServiceThread()
53	throws IllegalAccessException, InstantiationException, ClassNotFoundException {
54	try {
55	IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();	×
56	thriftServiceThread =	×
57	new ThriftServiceThread(
58	processor,
59	getID().getName(),	×
60	ThreadName.MLNODE_RPC_SERVICE.getName(),	×
61	getBindIP(),	×
62	getBindPort(),	×
63	config.getRpcMaxConcurrentClientNum(),	×
64	config.getThriftServerAwaitTimeForStopService(),	×
65	new MLNodeRPCServiceThriftHandler(impl),
66	false);
67	} catch (RPCServiceException e) {	×
68	throw new IllegalAccessException(e.getMessage());	×
69	}	×
70	thriftServiceThread.setName(ThreadName.MLNODE_RPC_SERVICE.getName());	×
71	}	×
72
73	@Override
74	public String getBindIP() {
75	return IoTDBDescriptor.getInstance().getConfig().getRpcAddress();	×
76	}
77
78	@Override
79	public int getBindPort() {
80	return IoTDBDescriptor.getInstance().getConfig().getMLNodePort();	×
81	}
82
83	private static class MLNodeRPCServiceHolder {
84	private static final MLNodeRPCService INSTANCE = new MLNodeRPCService();	×
85
86	private MLNodeRPCServiceHolder() {}
87	}
88
89	public static MLNodeRPCService getInstance() {
90	return MLNodeRPCService.MLNodeRPCServiceHolder.INSTANCE;	×
91	}
92	}

apache / iotdb / #10019

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous