RunIdGenerator { public Optional<String> getRunId() { DistributedLock runIdLock; String runId = null; runIdLock = coordinationUtils.getLock(CoordinationConstants.RUNID_LOCK_ID); if (runIdLock == null) { throw new SamzaException("Failed to create utils for run id generation"); } try { if (runIdLock.lock(Duration.ofMillis(CoordinationConstants.LOCK_TIMEOUT_MS))) { LOG.info("lock acquired for run.id generation by this processor"); processorId = clusterMembership.registerProcessor(); int numberOfActiveProcessors = clusterMembership.getNumberOfProcessors(); if (numberOfActiveProcessors == 0) { String msg = String.format("Processor failed to fetch number of processors for run.id generation"); throw new SamzaException(msg); } if (numberOfActiveProcessors == 1) { runId = String.valueOf(System.currentTimeMillis()) + "-" + UUID.randomUUID().toString().substring(0, 8); LOG.info("Writing the run id for this run as {}", runId); metadataStore.put(CoordinationConstants.RUNID_STORE_KEY, runId.getBytes("UTF-8")); metadataStore.flush(); } else { runId = new String(metadataStore.get(CoordinationConstants.RUNID_STORE_KEY)); LOG.info("Read the run id for this run as {}", runId); } runIdLock.unlock(); } else { throw new SamzaException("Processor timed out waiting to acquire lock for run.id generation"); } } catch (UnsupportedEncodingException e) { throw new SamzaException("Processor could not serialize/deserialize string for run.id generation", e); } return Optional.ofNullable(runId); } RunIdGenerator(CoordinationUtils coordinationUtils, MetadataStore metadataStore); Optional<String> getRunId(); void close();  }
@Test public void testSingleProcessorWriteRunId() throws Exception { prepareRunIdGenerator(1); runIdGenerator.getRunId(); verify(coordinationUtils, Mockito.times(1)).getClusterMembership(); verify(coordinationUtils, Mockito.times(1)).getLock(anyString()); verify(distributedLock, Mockito.times(1)).lock(anyObject()); verify(distributedLock, Mockito.times(1)).unlock(); verify(membership, Mockito.times(1)).registerProcessor(); verify(membership, Mockito.times(1)).getNumberOfProcessors(); verify(metadataStore, Mockito.times(1)).put(eq(CoordinationConstants.RUNID_STORE_KEY), any(byte[].class)); verify(metadataStore, Mockito.times(1)).flush(); }
@Test public void testTwoProcessorsReadRunId() throws Exception { prepareRunIdGenerator(2); String runId = runIdGenerator.getRunId().get(); assertEquals("Runid was not read from store", runId, FAKE_RUNID); verify(coordinationUtils, Mockito.times(1)).getClusterMembership(); verify(coordinationUtils, Mockito.times(1)).getLock(anyString()); verify(distributedLock, Mockito.times(1)).lock(anyObject()); verify(distributedLock, Mockito.times(1)).unlock(); verify(membership, Mockito.times(1)).registerProcessor(); verify(membership, Mockito.times(1)).getNumberOfProcessors(); verify(metadataStore, Mockito.times(1)).get(CoordinationConstants.RUNID_STORE_KEY); }