ZooKeeperManager { public void setViewInfo(ViewInfo viewInfo) throws Exception { String resource = viewInfo.getResource(); byte[] json = viewInfo.toCompressedJson(); LOG.info("Writtng compressed json for resource : " + resource + " - total bytes " + json.length); if (viewInfo.hasOnlinePartitions()) { String viewPath = this.clusterViewsPath + "/" + resource; if (zkClient.get().exists(viewPath, false) != null) { zkClient.get().setData(viewPath, json, -1); } else { zkClient.get().create(viewPath, json, ZK_ACL_LIST, CreateMode.PERSISTENT); } } else { LOG.info("No online partitions for " + resource + " - skipped writing."); } } ZooKeeperManager(ZooKeeperClient zkClient, String clusterName); void createClusterPaths(); void registerWatchAllFileSets(); void lockFileSet(String fileSet,
                          FileSetInfo fileSetInfo,
                          CreateMode mode); void lockFileSet(String fileSet,
                          FileSetInfo fileSetInfo); void unlockFileSet(String fileSet); void setFileSetInfo(String fileSet,
                             FileSetInfo fileSetInfo); void deleteFileSetInfo(String fileSet); FileSetInfo getFileSetInfo(String fileSet); ViewInfo getViewInfo(String resource); void setViewInfo(ViewInfo viewInfo); void deleteViewInfo(String resource); FileSetInfo getFileSetInfoForced(String fileSet); FileSetInfo getLockedFileSetInfo(String fileSet); Map<String, Pair<FileSetInfo, FileSetInfo>> getCandidateHdfsDirMap(); InetSocketAddress getControllerLeader(); ClusterInfo getClusterInfo(); void setClusterInfo(ClusterInfo info); Map<String, FileSetInfo> getFileSetInfoMap(); List<String> getLockedFileSets(); List<String> getLiveInstances();  }
@Test public void testSetExistingViewInfo() throws Exception { ViewInfo viewInfo = mock(ViewInfo.class); String resource = "test_resource"; String resourcePath = VIEWS_DIR + "/" + resource; byte[] resourceData = "hello".getBytes(); when(viewInfo.toCompressedJson()).thenReturn(resourceData); when(viewInfo.hasOnlinePartitions()).thenReturn(true); when(viewInfo.getResource()).thenReturn(resource); when(zk.exists(eq(resourcePath), anyBoolean())).thenReturn(new Stat()); when(zk.setData(eq(resourcePath), eq(resourceData), eq(-1))).thenReturn(new Stat()); zkManager.setViewInfo(viewInfo); verify(zk).setData(eq(resourcePath), eq(resourceData), eq(-1)); }
@Test public void testSetNonExistingViewInfo() throws Exception { ViewInfo viewInfo = mock(ViewInfo.class); String resource = "test_resource"; String resourcePath = VIEWS_DIR + "/" + resource; byte[] resourceData = "hello".getBytes(); when(viewInfo.toCompressedJson()).thenReturn(resourceData); when(viewInfo.hasOnlinePartitions()).thenReturn(true); when(viewInfo.getResource()).thenReturn(resource); when(zk.exists(eq(resourcePath), anyBoolean())).thenReturn(null); when(zk.create(eq(resourcePath), eq(resourceData), anyListOf(ACL.class), any(CreateMode.class))) .thenReturn(null); zkManager.setViewInfo(viewInfo); verify(zk).create(eq(resourcePath), eq(resourceData), anyListOf(ACL.class), any(CreateMode.class)); }
@Test public void testSetViewInfoNoOnlinePart() throws Exception { ViewInfo viewInfo = mock(ViewInfo.class); byte[] resourceData = "hello".getBytes(); when(viewInfo.toCompressedJson()).thenReturn(resourceData); when(viewInfo.hasOnlinePartitions()).thenReturn(false); zkManager.setViewInfo(viewInfo); verify(zk, times(0)).create(anyString(), any(byte[].class), anyListOf(ACL.class), any(CreateMode.class)); verify(zk, times(0)).setData(anyString(), any(byte[].class), anyInt()); }