### Question:
Etcd64bitIdGenerator implements LedgerIdGenerator { @Override public void generateLedgerId(GenericCallback<Long> cb) { int bucketId = nextBucketId(); checkArgument(bucketId >= 0 && bucketId < NUM_BUCKETS, "Invalid bucket id : " + bucketId); ByteSequence bucketKey = ByteSequence.fromString(EtcdUtils.getBucketPath(scope, bucketId)); Txn txn = kvClient.txn() .If(new Cmp(bucketKey, Op.GREATER, CmpTarget.createRevision(0))) .Then( com.coreos.jetcd.op.Op.put(bucketKey, EMPTY_BS, PutOption.DEFAULT), com.coreos.jetcd.op.Op.get(bucketKey, GetOption.DEFAULT) ) .Else( com.coreos.jetcd.op.Op.put(bucketKey, EMPTY_BS, PutOption.DEFAULT), com.coreos.jetcd.op.Op.get(bucketKey, GetOption.DEFAULT) ); txn.commit() .thenAccept(txnResponse -> { if (txnResponse.getGetResponses().size() <= 0) { cb.operationComplete(Code.UnexpectedConditionException, null); } else { GetResponse resp = txnResponse.getGetResponses().get(0); if (resp.getCount() > 0) { KeyValue kv = resp.getKvs().get(0); if (kv.getVersion() > MAX_ID_PER_BUCKET) { log.warn("Etcd bucket '{}' is overflowed", bucketKey.toStringUtf8()); generateLedgerId(cb); } else { long version = kv.getVersion(); long lid = ((((long) bucketId) << BUCKET_ID_SHIFT) & BUCKET_ID_MASK) | (version & MAX_ID_PER_BUCKET); cb.operationComplete(Code.OK, lid); } } else { cb.operationComplete(Code.UnexpectedConditionException, null); } } }) .exceptionally(cause -> { cb.operationComplete(Code.MetaStoreException, null); return null; }); } Etcd64bitIdGenerator(KV kvClient, String scope); @Override void generateLedgerId(GenericCallback<Long> cb); @Override void close();  }### Answer:
@Test public void testGenerateIdParallel() throws Exception { final int numThreads = 10; @Cleanup("shutdown") ExecutorService executor = Executors.newFixedThreadPool(numThreads); final int numIds = 10000; final AtomicLong totalIds = new AtomicLong(numIds); final Set<Long> ids = Collections.newSetFromMap(new ConcurrentHashMap<>()); final RateLimiter limiter = RateLimiter.create(1000); final CompletableFuture<Void> doneFuture = new CompletableFuture<>(); for (int i = 0; i < numThreads; i++) { executor.submit(() -> { Client client = Client.builder() .endpoints(etcdContainer.getClientEndpoint()) .build(); Etcd64bitIdGenerator gen = new Etcd64bitIdGenerator( client.getKVClient(), scope ); AtomicBoolean running = new AtomicBoolean(true); while (running.get()) { limiter.acquire(); GenericCallbackFuture<Long> genFuture = new GenericCallbackFuture<>(); gen.generateLedgerId(genFuture); genFuture .thenAccept(lid -> { boolean duplicatedFound = !(ids.add(lid)); if (duplicatedFound) { running.set(false); doneFuture.completeExceptionally( new IllegalStateException("Duplicated id " + lid + " generated : " + ids)); return; } else { if (totalIds.decrementAndGet() <= 0) { running.set(false); doneFuture.complete(null); } } }) .exceptionally(cause -> { running.set(false); doneFuture.completeExceptionally(cause); return null; }); } }); } FutureUtils.result(doneFuture); assertTrue(totalIds.get() <= 0); assertTrue(ids.size() >= numIds); }