### Question:
EtcdMetaStore implements MetaStore { @Override public void watch(WatchType type, LongConsumer consumer) { switch (type) { case SCHEMAVERSION: etcdClient.watch(SCHEMA_VERSION_PATH, false, consumer); break; case SCHEMATASK: etcdClient.watch(TASK_LIST, true, consumer); break; case PRIVERSION: etcdClient.watch(PRI_VERSION_PATH, false, consumer); break; case PRITASK: etcdClient.watch(TASK_PRI_LIST, true, consumer); break; case INDEXTASK: etcdClient.watch(TASK_INDEX_ROOT, true, consumer); break; default: break; } }  @Override void init(JimConfig c); @Override int allocMetaID(); @Override long allocTaskID(TaskType type); @Override long addAndGetVersion(long incr); @Override long addAndGetPrivVersion(long incr); @Override void register(long version); @Override void unRegister(); @Override Map<String, Long> getRegisters(); @Override boolean storeCatalog(Metapb.CatalogInfo expect, Metapb.CatalogInfo catalogInfo); @Override boolean removeCatalog(int id); @Override Metapb.CatalogInfo getCatalog(int id); @Override List<Metapb.CatalogInfo> getCatalogs(); @Override boolean storeTable(Metapb.TableInfo expect, Metapb.TableInfo tableInfo); @Override boolean removeTable(int dbID, int tableID); @Override Metapb.TableInfo getTable(int catalogID, int tableID); @Override List<Metapb.TableInfo> getTables(int catalogID); @Override Map<Metapb.CatalogInfo, List<Metapb.TableInfo>> getCatalogAndTables(); @Override Metapb.AutoIdInfo getAutoIdInfo(int catalogID, int tableID); @Override Metapb.AutoIdInfo storeAutoIdInfo(int catalogID, int tableID, Metapb.AutoIdInfo autoIdInfo, Metapb.AutoIdInfo expect); @Override boolean storeTask(TaskType type, Task expect, Task... tasks); @Override Task getTask(TaskType type, Task task); @Override List<Task> getTasks(TaskType type, Task task); @Override void removeTask(TaskType type, long taskId); @Override boolean tryLock(TaskType type, Task task); @Override void unLock(TaskType type, Task task); @Override void watch(WatchType type, LongConsumer consumer); @Override void close();  }### Answer:
@Test public void watchTest() throws InterruptedException { int parallel = 16; CountDownLatch latch = new CountDownLatch(parallel); ConcurrentHashMap<Integer, EtcdMetaStore> storeMap = new ConcurrentHashMap(); for (int i = 0; i < parallel; i++) { EtcdMetaStore metaStore = new EtcdMetaStore(); Properties properties = IOUtil.loadResource("jim.properties"); JimConfig config = new JimConfig(properties); metaStore.init(config); storeMap.put(i, metaStore); } storeMap.forEach((k, v) -> { Thread t = new Thread(() -> { v.watch(MetaStore.WatchType.SCHEMAVERSION, value -> Assert.assertEquals(value, 0)); latch.countDown(); }); t.start(); }); latch.await(); for (int i = 0; i < parallel; i++) { metaStore.addAndGetVersion(1); } TimeUnit.SECONDS.sleep(10); }