jt2594838 commented on a change in pull request #1464: URL: https://github.com/apache/incubator-iotdb/pull/1464#discussion_r450587628
########## File path: cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java ########## @@ -0,0 +1,233 @@ +/* + * 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.cluster.metadata; + +import org.apache.iotdb.db.conf.IoTDBConstant; +import org.apache.iotdb.db.exception.metadata.MetadataException; +import org.apache.iotdb.db.exception.metadata.PathNotExistException; +import org.apache.iotdb.db.metadata.MManager; +import org.apache.iotdb.db.metadata.MeasurementMeta; +import org.apache.iotdb.db.metadata.mnode.MNode; +import org.apache.iotdb.db.qp.physical.crud.InsertPlan; +import org.apache.iotdb.db.service.IoTDB; +import org.apache.iotdb.tsfile.common.cache.LRUCache; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.read.TimeValuePair; +import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +public class CMManager extends MManager { + + private static final Logger logger = LoggerFactory.getLogger(CMManager.class); + + // currently, if a key is not existed in the mRemoteMetaCache, an IOException will be thrown + private ReentrantReadWriteLock cacheLock = new ReentrantReadWriteLock();; + private LRUCache<String, MeasurementMeta> mRemoteMetaCache; + private MetaPuller metaPuller; + + protected CMManager() { + super(); + metaPuller = MetaPuller.getInstance(); + int remoteCacheSize = config.getmRemoteSchemaCacheSize(); + mRemoteMetaCache = new LRUCache<String, MeasurementMeta>(remoteCacheSize) { + @Override + protected MeasurementMeta loadObjectByKey(String key) throws IOException { + throw new IOException(key + " not found!"); + } + + @Override + public synchronized void removeItem(String key) { + cache.keySet().removeIf(s -> s.startsWith(key)); + } + }; + } + + private static class MManagerHolder { + + private MManagerHolder() { + // allowed to do nothing + } + + private static final CMManager INSTANCE = new CMManager(); + } + + /** + * we should not use this function in other place, but only in IoTDB class + * @return + */ + public static MManager getInstance() { + return CMManager.MManagerHolder.INSTANCE; + } + + @Override + public String deleteTimeseries(String prefixPath) throws MetadataException { + cacheLock.writeLock().lock(); + mRemoteMetaCache.removeItem(prefixPath); + cacheLock.writeLock().unlock(); + return super.deleteTimeseries(prefixPath); + } + + @Override + public void deleteStorageGroups(List<String> storageGroups) throws MetadataException { + cacheLock.writeLock().lock(); + for (String storageGroup : storageGroups) { + mRemoteMetaCache.removeItem(storageGroup); + } + cacheLock.writeLock().unlock(); + super.deleteStorageGroups(storageGroups); + } + + @Override + public TSDataType getSeriesType(String path) throws MetadataException { + try { + cacheLock.readLock().lock(); + MeasurementMeta measurementMeta = mRemoteMetaCache.get(path); + return measurementMeta.getMeasurementSchema().getType(); + } catch (IOException e) { + //do nothing Review comment: When is the schema pulled in this case? ########## File path: cluster/src/main/java/org/apache/iotdb/cluster/server/handlers/caller/AppendNodeEntryHandler.java ########## @@ -66,7 +67,7 @@ public void onComplete(Long response) { logger.debug("{}: Received an agreement from {} for {}, remaining votes to succeed: {}", member.getName(), receiver, log, remaining); if (remaining == 0) { - logger.debug("{}: Log {} is accepted by the quorum", member.getName(), log); + logger.debug("{}: Log {} is accepted by the quorum {}", member.getName(), log, log.getCurrLogIndex()); Review comment: Adding the parameter after "quorum" is a bit confusing, maybe we can use: `"{}: Log[{}] {} is accepted by the quorum", member.getName(), log.getCurrLogIndex(), log` ########## File path: cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java ########## @@ -0,0 +1,233 @@ +/* + * 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.cluster.metadata; + +import org.apache.iotdb.db.conf.IoTDBConstant; +import org.apache.iotdb.db.exception.metadata.MetadataException; +import org.apache.iotdb.db.exception.metadata.PathNotExistException; +import org.apache.iotdb.db.metadata.MManager; +import org.apache.iotdb.db.metadata.MeasurementMeta; +import org.apache.iotdb.db.metadata.mnode.MNode; +import org.apache.iotdb.db.qp.physical.crud.InsertPlan; +import org.apache.iotdb.db.service.IoTDB; +import org.apache.iotdb.tsfile.common.cache.LRUCache; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.read.TimeValuePair; +import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +public class CMManager extends MManager { + + private static final Logger logger = LoggerFactory.getLogger(CMManager.class); + + // currently, if a key is not existed in the mRemoteMetaCache, an IOException will be thrown + private ReentrantReadWriteLock cacheLock = new ReentrantReadWriteLock();; + private LRUCache<String, MeasurementMeta> mRemoteMetaCache; + private MetaPuller metaPuller; + + protected CMManager() { + super(); + metaPuller = MetaPuller.getInstance(); + int remoteCacheSize = config.getmRemoteSchemaCacheSize(); + mRemoteMetaCache = new LRUCache<String, MeasurementMeta>(remoteCacheSize) { + @Override + protected MeasurementMeta loadObjectByKey(String key) throws IOException { + throw new IOException(key + " not found!"); Review comment: Throwing an exception would fill the stack trace, so how about changing it to return null? ########## File path: cluster/src/main/java/org/apache/iotdb/cluster/metadata/MetaPuller.java ########## @@ -0,0 +1,182 @@ +/* + * 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.cluster.metadata; + +import org.apache.iotdb.cluster.client.async.AsyncDataClient; +import org.apache.iotdb.cluster.client.sync.SyncClientAdaptor; +import org.apache.iotdb.cluster.exception.CheckConsistencyException; +import org.apache.iotdb.cluster.partition.PartitionGroup; +import org.apache.iotdb.cluster.rpc.thrift.Node; +import org.apache.iotdb.cluster.rpc.thrift.PullSchemaRequest; +import org.apache.iotdb.cluster.server.member.MetaGroupMember; +import org.apache.iotdb.db.exception.metadata.MetadataException; +import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; +import org.apache.iotdb.db.service.IoTDB; +import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.apache.thrift.TException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class MetaPuller { + private static final Logger logger = LoggerFactory.getLogger(MetaPuller.class); + private MetaGroupMember metaGroupMember; + + private MetaPuller() { + } + + private static class MetaPullerHolder { + + private static final MetaPuller INSTANCE = new MetaPuller(); + } + + public void init(MetaGroupMember metaGroupMember) { + this.metaGroupMember = metaGroupMember; + } + + /** + * we should not use this function in other place, but only in IoTDB class + * @return + */ + public static MetaPuller getInstance() { + return MetaPullerHolder.INSTANCE; + } + + /** + * Pull the all timeseries schemas of given prefixPaths from remote nodes. All prefixPaths must + * contain the storage group. + */ + public List<MeasurementSchema> pullTimeSeriesSchemas(List<String> prefixPaths) + throws MetadataException { + logger.debug("{}: Pulling timeseries schemas of {}", metaGroupMember.getName(), prefixPaths); + // split the paths by the data groups that will hold them + Map<PartitionGroup, List<String>> partitionGroupPathMap = new HashMap<>(); + for (String prefixPath : prefixPaths) { + PartitionGroup partitionGroup; + try { + partitionGroup = metaGroupMember.getPartitionTable().partitionByPathTime(prefixPath, 0); + } catch (StorageGroupNotSetException e) { + // the storage group is not found locally, but may be found in the leader, retry after + // synchronizing with the leader + + try { + metaGroupMember.syncLeaderWithConsistencyCheck(); + } catch (CheckConsistencyException checkConsistencyException) { + throw new MetadataException(checkConsistencyException.getMessage()); + } + partitionGroup = metaGroupMember.getPartitionTable().partitionByPathTime(prefixPath, 0); + + } + partitionGroupPathMap.computeIfAbsent(partitionGroup, g -> new ArrayList<>()).add(prefixPath); + } + + List<MeasurementSchema> schemas = new ArrayList<>(); + // pull timeseries schema from every group involved + if (logger.isDebugEnabled()) { + logger.debug("{}: pulling schemas of {} and other {} paths from {} groups", metaGroupMember.getName(), + prefixPaths.get(0), prefixPaths.size() - 1, + partitionGroupPathMap.size()); + } + for (Map.Entry<PartitionGroup, List<String>> partitionGroupListEntry : partitionGroupPathMap + .entrySet()) { + PartitionGroup partitionGroup = partitionGroupListEntry.getKey(); + List<String> paths = partitionGroupListEntry.getValue(); + pullTimeSeriesSchemas(partitionGroup, paths, schemas); + } Review comment: How about using parallel pulling? ########## File path: cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java ########## @@ -0,0 +1,233 @@ +/* + * 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.cluster.metadata; + +import org.apache.iotdb.db.conf.IoTDBConstant; +import org.apache.iotdb.db.exception.metadata.MetadataException; +import org.apache.iotdb.db.exception.metadata.PathNotExistException; +import org.apache.iotdb.db.metadata.MManager; +import org.apache.iotdb.db.metadata.MeasurementMeta; +import org.apache.iotdb.db.metadata.mnode.MNode; +import org.apache.iotdb.db.qp.physical.crud.InsertPlan; +import org.apache.iotdb.db.service.IoTDB; +import org.apache.iotdb.tsfile.common.cache.LRUCache; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.read.TimeValuePair; +import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +public class CMManager extends MManager { + + private static final Logger logger = LoggerFactory.getLogger(CMManager.class); + + // currently, if a key is not existed in the mRemoteMetaCache, an IOException will be thrown + private ReentrantReadWriteLock cacheLock = new ReentrantReadWriteLock();; + private LRUCache<String, MeasurementMeta> mRemoteMetaCache; + private MetaPuller metaPuller; + + protected CMManager() { + super(); + metaPuller = MetaPuller.getInstance(); + int remoteCacheSize = config.getmRemoteSchemaCacheSize(); + mRemoteMetaCache = new LRUCache<String, MeasurementMeta>(remoteCacheSize) { + @Override + protected MeasurementMeta loadObjectByKey(String key) throws IOException { + throw new IOException(key + " not found!"); + } + + @Override + public synchronized void removeItem(String key) { + cache.keySet().removeIf(s -> s.startsWith(key)); + } + }; + } + + private static class MManagerHolder { + + private MManagerHolder() { + // allowed to do nothing + } + + private static final CMManager INSTANCE = new CMManager(); + } + + /** + * we should not use this function in other place, but only in IoTDB class + * @return + */ + public static MManager getInstance() { + return CMManager.MManagerHolder.INSTANCE; + } + + @Override + public String deleteTimeseries(String prefixPath) throws MetadataException { + cacheLock.writeLock().lock(); + mRemoteMetaCache.removeItem(prefixPath); + cacheLock.writeLock().unlock(); + return super.deleteTimeseries(prefixPath); + } + + @Override + public void deleteStorageGroups(List<String> storageGroups) throws MetadataException { + cacheLock.writeLock().lock(); + for (String storageGroup : storageGroups) { + mRemoteMetaCache.removeItem(storageGroup); + } + cacheLock.writeLock().unlock(); + super.deleteStorageGroups(storageGroups); + } + + @Override + public TSDataType getSeriesType(String path) throws MetadataException { + try { + cacheLock.readLock().lock(); + MeasurementMeta measurementMeta = mRemoteMetaCache.get(path); + return measurementMeta.getMeasurementSchema().getType(); + } catch (IOException e) { + //do nothing + } finally { + cacheLock.readLock().unlock(); + } + return super.getSeriesType(path); + } + + @Override + public MeasurementSchema[] getSchemas(String deviceId, String[] measurements) throws MetadataException { + try { + return super.getSchemas(deviceId, measurements); + } catch (MetadataException e) { + // some measurements not exist in local + // try cache + MeasurementSchema[] measurementSchemas = new MeasurementSchema[measurements.length]; + boolean allSeriesExists = true; + cacheLock.readLock().lock(); + for (int i = 0; i < measurements.length; i++) { + try { + MeasurementMeta measurementMeta = mRemoteMetaCache.get(deviceId + measurements[i]); + measurementSchemas[i] = measurementMeta.getMeasurementSchema(); + } catch (IOException ex) { + // not all cached, pull from remote + allSeriesExists = false; + break; + } + } + cacheLock.readLock().unlock(); + if (allSeriesExists) { + return measurementSchemas; + } + + pullSeriesSchemas(deviceId, measurements); + + // try again + boolean allExist = true; + cacheLock.readLock().lock(); + for (int i = 0; i < measurements.length; i++) { + try { + MeasurementMeta measurementMeta = mRemoteMetaCache.get(deviceId + measurements[i]); + measurementSchemas[i] = measurementMeta.getMeasurementSchema(); + } catch (IOException ex) { + allExist = false; + break; + } + } + cacheLock.readLock().unlock(); + if (!allExist) { + throw new MetadataException(deviceId + " has some mesurements not found"); Review comment: It would be better to show which series is not found and just throw PathNotExistException. ########## File path: cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java ########## @@ -0,0 +1,233 @@ +/* + * 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.cluster.metadata; + +import org.apache.iotdb.db.conf.IoTDBConstant; +import org.apache.iotdb.db.exception.metadata.MetadataException; +import org.apache.iotdb.db.exception.metadata.PathNotExistException; +import org.apache.iotdb.db.metadata.MManager; +import org.apache.iotdb.db.metadata.MeasurementMeta; +import org.apache.iotdb.db.metadata.mnode.MNode; +import org.apache.iotdb.db.qp.physical.crud.InsertPlan; +import org.apache.iotdb.db.service.IoTDB; +import org.apache.iotdb.tsfile.common.cache.LRUCache; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.read.TimeValuePair; +import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +public class CMManager extends MManager { + + private static final Logger logger = LoggerFactory.getLogger(CMManager.class); + + // currently, if a key is not existed in the mRemoteMetaCache, an IOException will be thrown + private ReentrantReadWriteLock cacheLock = new ReentrantReadWriteLock();; + private LRUCache<String, MeasurementMeta> mRemoteMetaCache; + private MetaPuller metaPuller; + + protected CMManager() { + super(); + metaPuller = MetaPuller.getInstance(); + int remoteCacheSize = config.getmRemoteSchemaCacheSize(); + mRemoteMetaCache = new LRUCache<String, MeasurementMeta>(remoteCacheSize) { + @Override + protected MeasurementMeta loadObjectByKey(String key) throws IOException { + throw new IOException(key + " not found!"); + } + + @Override + public synchronized void removeItem(String key) { + cache.keySet().removeIf(s -> s.startsWith(key)); + } + }; + } + + private static class MManagerHolder { + + private MManagerHolder() { + // allowed to do nothing + } + + private static final CMManager INSTANCE = new CMManager(); + } + + /** + * we should not use this function in other place, but only in IoTDB class + * @return + */ + public static MManager getInstance() { + return CMManager.MManagerHolder.INSTANCE; + } + + @Override + public String deleteTimeseries(String prefixPath) throws MetadataException { + cacheLock.writeLock().lock(); + mRemoteMetaCache.removeItem(prefixPath); + cacheLock.writeLock().unlock(); + return super.deleteTimeseries(prefixPath); + } + + @Override + public void deleteStorageGroups(List<String> storageGroups) throws MetadataException { + cacheLock.writeLock().lock(); + for (String storageGroup : storageGroups) { + mRemoteMetaCache.removeItem(storageGroup); + } + cacheLock.writeLock().unlock(); + super.deleteStorageGroups(storageGroups); + } + + @Override + public TSDataType getSeriesType(String path) throws MetadataException { + try { + cacheLock.readLock().lock(); + MeasurementMeta measurementMeta = mRemoteMetaCache.get(path); + return measurementMeta.getMeasurementSchema().getType(); + } catch (IOException e) { + //do nothing + } finally { + cacheLock.readLock().unlock(); + } + return super.getSeriesType(path); + } + + @Override + public MeasurementSchema[] getSchemas(String deviceId, String[] measurements) throws MetadataException { + try { + return super.getSchemas(deviceId, measurements); + } catch (MetadataException e) { + // some measurements not exist in local + // try cache + MeasurementSchema[] measurementSchemas = new MeasurementSchema[measurements.length]; + boolean allSeriesExists = true; + cacheLock.readLock().lock(); + for (int i = 0; i < measurements.length; i++) { + try { + MeasurementMeta measurementMeta = mRemoteMetaCache.get(deviceId + measurements[i]); + measurementSchemas[i] = measurementMeta.getMeasurementSchema(); + } catch (IOException ex) { + // not all cached, pull from remote + allSeriesExists = false; + break; + } + } + cacheLock.readLock().unlock(); + if (allSeriesExists) { + return measurementSchemas; + } + + pullSeriesSchemas(deviceId, measurements); + + // try again + boolean allExist = true; + cacheLock.readLock().lock(); + for (int i = 0; i < measurements.length; i++) { + try { + MeasurementMeta measurementMeta = mRemoteMetaCache.get(deviceId + measurements[i]); + measurementSchemas[i] = measurementMeta.getMeasurementSchema(); + } catch (IOException ex) { + allExist = false; + break; + } + } + cacheLock.readLock().unlock(); + if (!allExist) { + throw new MetadataException(deviceId + " has some mesurements not found"); + } + return measurementSchemas; + } + } + + private void pullSeriesSchemas(String deviceId, String[] measurementList) + throws MetadataException { + List<String> schemasToPull = new ArrayList<>(); + for (String s : measurementList) { + schemasToPull.add(deviceId + IoTDBConstant.PATH_SEPARATOR + s); + } + List<MeasurementSchema> schemas = metaPuller.pullTimeSeriesSchemas(schemasToPull); + for (MeasurementSchema schema : schemas) { + cacheMeta(deviceId + IoTDBConstant.PATH_SEPARATOR + schema.getMeasurementId(), new MeasurementMeta(schema)); + } + logger.debug("Pulled {}/{} schemas from remote", schemas.size(), measurementList.length); + } + + @Override + public void cacheMeta(String seriesPath, MeasurementMeta meta) { + cacheLock.writeLock().lock(); + mRemoteMetaCache.put(seriesPath, meta); + cacheLock.writeLock().unlock(); + } + + @Override + public void updateLastCache(String seriesPath, TimeValuePair timeValuePair, boolean highPriorityUpdate, Long latestFlushedTime) { + cacheLock.writeLock().lock(); + try { + MeasurementMeta measurementMeta = mRemoteMetaCache.get(seriesPath); + measurementMeta.updateCachedLast(timeValuePair, highPriorityUpdate, latestFlushedTime); + } catch (IOException e) { + // not found + } finally { + cacheLock.writeLock().unlock(); + } + // maybe local also has the timeseries + super.updateLastCache(seriesPath, timeValuePair, highPriorityUpdate, latestFlushedTime); + } + + @Override + public TimeValuePair getLastCache(String seriesPath) { + try { + MeasurementMeta measurementMeta = mRemoteMetaCache.get(seriesPath); + return measurementMeta.getTimeValuePair(); + } catch (IOException e) { + // do nothing + } + return super.getLastCache(seriesPath); + } + + @Override + public MeasurementSchema[] getSeriesSchemasAndReadLockDevice(String deviceId, String[] measurementList, InsertPlan plan) throws MetadataException { + //TODO cluster also need to lock device node Review comment: I think this is urgent, or you can extract the lock method separately. Otherwise, an IllegalMonitorStateException will be thrown because the next step will try to unlock but it is not locked. ########## File path: cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java ########## @@ -184,15 +182,15 @@ public void testApplyInsert() assertFalse(dataSet.hasNext()); // this series is not created but can be fetched - insertPlan.setDeviceId(TestUtils.getTestSg(4)); - applier.apply(log); - dataSet = query(Collections.singletonList(TestUtils.getTestSeries(4, 0)), null); - assertTrue(dataSet.hasNext()); - record = dataSet.next(); - assertEquals(1, record.getTimestamp()); - assertEquals(1, record.getFields().size()); - assertEquals(1.0, record.getFields().get(0).getDoubleV(), 0.00001); - assertFalse(dataSet.hasNext()); +// insertPlan.setDeviceId(TestUtils.getTestSg(4)); +// applier.apply(log); +// dataSet = query(Collections.singletonList(TestUtils.getTestSeries(4, 0)), null); +// assertTrue(dataSet.hasNext()); +// record = dataSet.next(); +// assertEquals(1, record.getTimestamp()); +// assertEquals(1, record.getFields().size()); +// assertEquals(1.0, record.getFields().get(0).getDoubleV(), 0.00001); +// assertFalse(dataSet.hasNext()); Review comment: Why are these commented? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
