SetupSteps { @PostConstruct public void runSetup() throws SetupException { HAConfiguration.ZookeeperProperties zookeeperProperties = HAConfiguration.getZookeeperProperties(configuration); InterProcessMutex lock = curatorFactory.lockInstance(zookeeperProperties.getZkRoot()); try { LOG.info("Trying to acquire lock for running setup."); lock.acquire(); LOG.info("Acquired lock for running setup."); handleSetupInProgress(configuration, zookeeperProperties); for (SetupStep step : setupSteps) { LOG.info("Running setup step: {}", step); step.run(); } clearSetupInProgress(zookeeperProperties); } catch (SetupException se) { LOG.error("Got setup exception while trying to setup", se); throw se; } catch (Throwable e) { LOG.error("Error running setup steps", e); throw new SetupException("Error running setup steps", e); } finally { releaseLock(lock); curatorFactory.close(); } } @Inject  SetupSteps(Set<SetupStep> steps, CuratorFactory curatorFactory, Configuration configuration); @PostConstruct void runSetup(); static final String SETUP_IN_PROGRESS_NODE; }
@Test public void shouldRunRegisteredSetupSteps() throws Exception { Set<SetupStep> steps = new LinkedHashSet<>(); SetupStep setupStep1 = mock(SetupStep.class); SetupStep setupStep2 = mock(SetupStep.class); steps.add(setupStep1); steps.add(setupStep2); when(configuration. getString(HAConfiguration.ATLAS_SERVER_HA_ZK_ROOT_KEY, HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT); setupServerIdSelectionMocks(); setupSetupInProgressPathMocks(ZooDefs.Ids.OPEN_ACL_UNSAFE); InterProcessMutex lock = mock(InterProcessMutex.class); when(curatorFactory.lockInstance(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(lock); SetupSteps setupSteps = new SetupSteps(steps, curatorFactory, configuration); setupSteps.runSetup(); verify(setupStep1).run(); verify(setupStep2).run(); }
@Test public void shouldRunSetupStepsUnderLock() throws Exception { Set<SetupStep> steps = new LinkedHashSet<>(); SetupStep setupStep1 = mock(SetupStep.class); SetupStep setupStep2 = mock(SetupStep.class); steps.add(setupStep1); steps.add(setupStep2); when(configuration. getString(HAConfiguration.ATLAS_SERVER_HA_ZK_ROOT_KEY, HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT); setupServerIdSelectionMocks(); setupSetupInProgressPathMocks(ZooDefs.Ids.OPEN_ACL_UNSAFE); InterProcessMutex lock = mock(InterProcessMutex.class); when(curatorFactory.lockInstance(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(lock); InOrder inOrder = inOrder(lock, setupStep1, setupStep2); SetupSteps setupSteps = new SetupSteps(steps, curatorFactory, configuration); setupSteps.runSetup(); inOrder.verify(lock).acquire(); inOrder.verify(setupStep1).run(); inOrder.verify(setupStep2).run(); inOrder.verify(lock).release(); }
@Test public void shouldReleaseLockOnException() throws Exception { Set<SetupStep> steps = new LinkedHashSet<>(); SetupStep setupStep1 = mock(SetupStep.class); steps.add(setupStep1); when(configuration. getString(HAConfiguration.ATLAS_SERVER_HA_ZK_ROOT_KEY, HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT); setupServerIdSelectionMocks(); setupSetupInProgressPathMocks(ZooDefs.Ids.OPEN_ACL_UNSAFE); doThrow(new RuntimeException("Simulating setup failure.")).when(setupStep1).run(); InterProcessMutex lock = mock(InterProcessMutex.class); when(curatorFactory.lockInstance(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(lock); InOrder inOrder = inOrder(lock, setupStep1); SetupSteps setupSteps = new SetupSteps(steps, curatorFactory, configuration); try { setupSteps.runSetup(); } catch (Exception e) { assertTrue(e instanceof SetupException); } inOrder.verify(lock).acquire(); inOrder.verify(setupStep1).run(); inOrder.verify(lock).release(); }
@Test public void shouldCreateSetupInProgressNode() throws Exception { Set<SetupStep> steps = new LinkedHashSet<>(); SetupStep setupStep1 = mock(SetupStep.class); steps.add(setupStep1); when(configuration. getString(HAConfiguration.ATLAS_SERVER_HA_ZK_ROOT_KEY, HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT); when(configuration.getString(HAConfiguration.HA_ZOOKEEPER_ACL)).thenReturn("digest:user:pwd"); List<ACL> aclList = Arrays.asList(new ACL(ZooDefs.Perms.ALL, new Id("digest", "user:pwd"))); setupServerIdSelectionMocks(); CreateBuilder createBuilder = setupSetupInProgressPathMocks(aclList).getLeft(); InterProcessMutex lock = mock(InterProcessMutex.class); when(curatorFactory.lockInstance(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(lock); SetupSteps setupSteps = new SetupSteps(steps, curatorFactory, configuration); setupSteps.runSetup(); verify(createBuilder).withACL(aclList); verify(createBuilder).forPath(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT+SetupSteps.SETUP_IN_PROGRESS_NODE, "id2".getBytes(Charsets.UTF_8)); }
@Test public void shouldDeleteSetupInProgressNodeAfterCompletion() throws Exception { Set<SetupStep> steps = new LinkedHashSet<>(); SetupStep setupStep1 = mock(SetupStep.class); steps.add(setupStep1); when(configuration. getString(HAConfiguration.ATLAS_SERVER_HA_ZK_ROOT_KEY, HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT); when(configuration.getString(HAConfiguration.HA_ZOOKEEPER_ACL)).thenReturn("digest:user:pwd"); List<ACL> aclList = Arrays.asList(new ACL(ZooDefs.Perms.ALL, new Id("digest", "user:pwd"))); setupServerIdSelectionMocks(); DeleteBuilder deleteBuilder = setupSetupInProgressPathMocks(aclList).getRight(); InterProcessMutex lock = mock(InterProcessMutex.class); when(curatorFactory.lockInstance(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(lock); SetupSteps setupSteps = new SetupSteps(steps, curatorFactory, configuration); setupSteps.runSetup(); verify(deleteBuilder).forPath(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT+SetupSteps.SETUP_IN_PROGRESS_NODE); }
@Test public void shouldThrowSetupExceptionAndNotDoSetupIfSetupInProgressNodeExists() throws Exception { Set<SetupStep> steps = new LinkedHashSet<>(); SetupStep setupStep1 = mock(SetupStep.class); steps.add(setupStep1); when(configuration. getString(HAConfiguration.ATLAS_SERVER_HA_ZK_ROOT_KEY, HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT); setupServerIdSelectionMocks(); setupSetupInProgressPathMocks(ZooDefs.Ids.OPEN_ACL_UNSAFE, mock(Stat.class)); InterProcessMutex lock = mock(InterProcessMutex.class); when(curatorFactory.lockInstance(HAConfiguration.ATLAS_SERVER_ZK_ROOT_DEFAULT)). thenReturn(lock); SetupSteps setupSteps = new SetupSteps(steps, curatorFactory, configuration); try { setupSteps.runSetup(); } catch (Exception e) { assertTrue(e instanceof SetupException); } verifyZeroInteractions(setupStep1); }