| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.distributedlog.impl.metadata; |
| |
| import static com.google.common.base.Charsets.UTF_8; |
| import static com.google.common.base.Preconditions.checkArgument; |
| import static com.google.common.base.Preconditions.checkNotNull; |
| import static org.apache.distributedlog.DistributedLogConstants.EMPTY_BYTES; |
| import static org.apache.distributedlog.DistributedLogConstants.UNASSIGNED_LOGSEGMENT_SEQNO; |
| import static org.apache.distributedlog.metadata.LogMetadata.*; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Optional; |
| import com.google.common.collect.Lists; |
| import java.io.IOException; |
| import java.net.URI; |
| import java.util.Collections; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.concurrent.CancellationException; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.TimeUnit; |
| import org.apache.bookkeeper.common.concurrent.FutureEventListener; |
| import org.apache.bookkeeper.stats.StatsLogger; |
| import org.apache.bookkeeper.versioning.LongVersion; |
| import org.apache.bookkeeper.versioning.Versioned; |
| import org.apache.distributedlog.DistributedLogConfiguration; |
| import org.apache.distributedlog.DistributedLogConstants; |
| import org.apache.distributedlog.LogSegmentMetadata; |
| import org.apache.distributedlog.ZooKeeperClient; |
| import org.apache.distributedlog.ZooKeeperClient.ZooKeeperConnectionException; |
| import org.apache.distributedlog.common.concurrent.FutureUtils; |
| import org.apache.distributedlog.common.util.PermitManager; |
| import org.apache.distributedlog.common.util.SchedulerUtils; |
| import org.apache.distributedlog.exceptions.DLInterruptedException; |
| import org.apache.distributedlog.exceptions.InvalidStreamNameException; |
| import org.apache.distributedlog.exceptions.LockCancelledException; |
| import org.apache.distributedlog.exceptions.LockingException; |
| import org.apache.distributedlog.exceptions.LogExistsException; |
| import org.apache.distributedlog.exceptions.LogNotFoundException; |
| import org.apache.distributedlog.exceptions.UnexpectedException; |
| import org.apache.distributedlog.exceptions.ZKException; |
| import org.apache.distributedlog.impl.ZKLogSegmentMetadataStore; |
| import org.apache.distributedlog.lock.DistributedLock; |
| import org.apache.distributedlog.lock.SessionLockFactory; |
| import org.apache.distributedlog.lock.ZKDistributedLock; |
| import org.apache.distributedlog.lock.ZKSessionLockFactory; |
| import org.apache.distributedlog.logsegment.LogSegmentMetadataStore; |
| import org.apache.distributedlog.metadata.LogMetadata; |
| import org.apache.distributedlog.metadata.LogMetadataForReader; |
| import org.apache.distributedlog.metadata.LogMetadataForWriter; |
| import org.apache.distributedlog.metadata.LogStreamMetadataStore; |
| import org.apache.distributedlog.util.DLUtils; |
| import org.apache.distributedlog.util.OrderedScheduler; |
| import org.apache.distributedlog.util.Transaction; |
| import org.apache.distributedlog.util.Utils; |
| import org.apache.distributedlog.zk.LimitedPermitManager; |
| import org.apache.distributedlog.zk.ZKTransaction; |
| import org.apache.zookeeper.AsyncCallback; |
| import org.apache.zookeeper.CreateMode; |
| import org.apache.zookeeper.KeeperException; |
| import org.apache.zookeeper.KeeperException.Code; |
| import org.apache.zookeeper.Op; |
| import org.apache.zookeeper.Op.Create; |
| import org.apache.zookeeper.Op.Delete; |
| import org.apache.zookeeper.OpResult; |
| import org.apache.zookeeper.ZKUtil; |
| import org.apache.zookeeper.ZooKeeper; |
| import org.apache.zookeeper.common.PathUtils; |
| import org.apache.zookeeper.data.ACL; |
| import org.apache.zookeeper.data.Stat; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * zookeeper based {@link LogStreamMetadataStore}. |
| */ |
| public class ZKLogStreamMetadataStore implements LogStreamMetadataStore { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(ZKLogStreamMetadataStore.class); |
| |
| private final String clientId; |
| private final DistributedLogConfiguration conf; |
| private final ZooKeeperClient zooKeeperClient; |
| private final OrderedScheduler scheduler; |
| private final StatsLogger statsLogger; |
| private final LogSegmentMetadataStore logSegmentStore; |
| private final LimitedPermitManager permitManager; |
| // lock |
| private SessionLockFactory lockFactory; |
| private OrderedScheduler lockStateExecutor; |
| |
| public ZKLogStreamMetadataStore(String clientId, |
| DistributedLogConfiguration conf, |
| ZooKeeperClient zkc, |
| OrderedScheduler scheduler, |
| StatsLogger statsLogger) { |
| this.clientId = clientId; |
| this.conf = conf; |
| this.zooKeeperClient = zkc; |
| this.scheduler = scheduler; |
| this.statsLogger = statsLogger; |
| // create the log segment metadata store and the permit manager (used for log segment rolling) |
| this.logSegmentStore = new ZKLogSegmentMetadataStore(conf, zooKeeperClient, scheduler); |
| this.permitManager = new LimitedPermitManager( |
| conf.getLogSegmentRollingConcurrency(), |
| 1, |
| TimeUnit.MINUTES, |
| scheduler); |
| this.zooKeeperClient.register(permitManager); |
| } |
| |
| private synchronized OrderedScheduler getLockStateExecutor(boolean createIfNull) { |
| if (createIfNull && null == lockStateExecutor) { |
| lockStateExecutor = OrderedScheduler.newBuilder() |
| .name("DLM-LockState") |
| .corePoolSize(conf.getNumLockStateThreads()) |
| .build(); |
| } |
| return lockStateExecutor; |
| } |
| |
| private synchronized SessionLockFactory getLockFactory(boolean createIfNull) { |
| if (createIfNull && null == lockFactory) { |
| lockFactory = new ZKSessionLockFactory( |
| zooKeeperClient, |
| clientId, |
| getLockStateExecutor(createIfNull), |
| conf.getZKNumRetries(), |
| conf.getLockTimeoutMilliSeconds(), |
| conf.getZKRetryBackoffStartMillis(), |
| statsLogger); |
| } |
| return lockFactory; |
| } |
| |
| @Override |
| public void close() throws IOException { |
| this.zooKeeperClient.unregister(permitManager); |
| this.permitManager.close(); |
| this.logSegmentStore.close(); |
| SchedulerUtils.shutdownScheduler( |
| getLockStateExecutor(false), |
| conf.getSchedulerShutdownTimeoutMs(), |
| TimeUnit.MILLISECONDS); |
| } |
| |
| @Override |
| public LogSegmentMetadataStore getLogSegmentMetadataStore() { |
| return logSegmentStore; |
| } |
| |
| @Override |
| public PermitManager getPermitManager() { |
| return this.permitManager; |
| } |
| |
| @Override |
| public Transaction<Object> newTransaction() { |
| return new ZKTransaction(zooKeeperClient); |
| } |
| |
| @Override |
| public CompletableFuture<Void> logExists(URI uri, final String logName) { |
| final String logSegmentsPath = LogMetadata.getLogSegmentsPath( |
| uri, logName, conf.getUnpartitionedStreamName()); |
| final CompletableFuture<Void> promise = new CompletableFuture<Void>(); |
| try { |
| final ZooKeeper zk = zooKeeperClient.get(); |
| zk.sync(logSegmentsPath, new AsyncCallback.VoidCallback() { |
| @Override |
| public void processResult(int syncRc, String path, Object syncCtx) { |
| if (KeeperException.Code.NONODE.intValue() == syncRc) { |
| promise.completeExceptionally(new LogNotFoundException( |
| String.format("Log %s does not exist or has been deleted", logName))); |
| return; |
| } else if (KeeperException.Code.OK.intValue() != syncRc){ |
| promise.completeExceptionally(new ZKException("Error on checking log existence for " + logName, |
| KeeperException.create(KeeperException.Code.get(syncRc)))); |
| return; |
| } |
| zk.exists(logSegmentsPath, false, new AsyncCallback.StatCallback() { |
| @Override |
| public void processResult(int rc, String path, Object ctx, Stat stat) { |
| if (KeeperException.Code.OK.intValue() == rc) { |
| promise.complete(null); |
| } else if (KeeperException.Code.NONODE.intValue() == rc) { |
| promise.completeExceptionally(new LogNotFoundException( |
| String.format("Log %s does not exist or has been deleted", logName))); |
| } else { |
| promise.completeExceptionally( |
| new ZKException("Error on checking log existence for " |
| + logName, KeeperException.create(KeeperException.Code.get(rc)))); |
| } |
| } |
| }, null); |
| } |
| }, null); |
| |
| } catch (InterruptedException ie) { |
| LOG.error("Interrupted while reading {}", logSegmentsPath, ie); |
| promise.completeExceptionally(new DLInterruptedException("Interrupted while checking " |
| + logSegmentsPath, ie)); |
| } catch (ZooKeeperClient.ZooKeeperConnectionException e) { |
| promise.completeExceptionally(e); |
| } |
| return promise; |
| } |
| |
| // |
| // Create Write Lock |
| // |
| |
| @Override |
| public DistributedLock createWriteLock(LogMetadataForWriter metadata) { |
| return new ZKDistributedLock( |
| getLockStateExecutor(true), |
| getLockFactory(true), |
| metadata.getLockPath(), |
| conf.getLockTimeoutMilliSeconds(), |
| statsLogger); |
| } |
| |
| // |
| // Create Read Lock |
| // |
| |
| private CompletableFuture<Void> ensureReadLockPathExist(final LogMetadata logMetadata, |
| final String readLockPath) { |
| final CompletableFuture<Void> promise = new CompletableFuture<Void>(); |
| promise.whenComplete((value, cause) -> { |
| if (cause instanceof CancellationException) { |
| FutureUtils.completeExceptionally(promise, new LockCancelledException(readLockPath, |
| "Could not ensure read lock path", cause)); |
| } |
| }); |
| Optional<String> parentPathShouldNotCreate = Optional.of(logMetadata.getLogRootPath()); |
| Utils.zkAsyncCreateFullPathOptimisticRecursive(zooKeeperClient, readLockPath, parentPathShouldNotCreate, |
| new byte[0], zooKeeperClient.getDefaultACL(), CreateMode.PERSISTENT, |
| new org.apache.zookeeper.AsyncCallback.StringCallback() { |
| @Override |
| public void processResult(final int rc, final String path, Object ctx, String name) { |
| if (KeeperException.Code.NONODE.intValue() == rc) { |
| FutureUtils.completeExceptionally(promise, new LogNotFoundException( |
| String.format("Log %s does not exist or has been deleted", |
| logMetadata.getFullyQualifiedName()))); |
| } else if (KeeperException.Code.OK.intValue() == rc) { |
| FutureUtils.complete(promise, null); |
| LOG.trace("Created path {}.", path); |
| } else if (KeeperException.Code.NODEEXISTS.intValue() == rc) { |
| FutureUtils.complete(promise, null); |
| LOG.trace("Path {} is already existed.", path); |
| } else if (DistributedLogConstants.ZK_CONNECTION_EXCEPTION_RESULT_CODE == rc) { |
| FutureUtils.completeExceptionally(promise, |
| new ZooKeeperClient.ZooKeeperConnectionException(path)); |
| } else if (DistributedLogConstants.DL_INTERRUPTED_EXCEPTION_RESULT_CODE == rc) { |
| FutureUtils.completeExceptionally(promise, new DLInterruptedException(path)); |
| } else { |
| FutureUtils.completeExceptionally(promise, |
| KeeperException.create(KeeperException.Code.get(rc))); |
| } |
| } |
| }, null); |
| return promise; |
| } |
| |
| @Override |
| public CompletableFuture<DistributedLock> createReadLock(final LogMetadataForReader metadata, |
| Optional<String> readerId) { |
| final String readLockPath = metadata.getReadLockPath(readerId); |
| return ensureReadLockPathExist(metadata, readLockPath) |
| .thenApplyAsync((value) -> { |
| DistributedLock lock = new ZKDistributedLock( |
| getLockStateExecutor(true), |
| getLockFactory(true), |
| readLockPath, |
| conf.getLockTimeoutMilliSeconds(), |
| statsLogger.scope("read_lock")); |
| return lock; |
| }, scheduler.chooseExecutor(readLockPath)); |
| } |
| |
| // |
| // Create Log |
| // |
| |
| static class MetadataIndex { |
| static final int LOG_ROOT_PARENT = 0; |
| static final int LOG_ROOT = 1; |
| static final int MAX_TXID = 2; |
| static final int VERSION = 3; |
| static final int LOCK = 4; |
| static final int READ_LOCK = 5; |
| static final int LOGSEGMENTS = 6; |
| static final int ALLOCATION = 7; |
| } |
| |
| static int bytesToInt(byte[] b) { |
| assert b.length >= 4; |
| return b[0] << 24 | b[1] << 16 | b[2] << 8 | b[3]; |
| } |
| |
| static byte[] intToBytes(int i) { |
| return new byte[]{ |
| (byte) (i >> 24), |
| (byte) (i >> 16), |
| (byte) (i >> 8), |
| (byte) (i)}; |
| } |
| |
| static CompletableFuture<List<Versioned<byte[]>>> checkLogMetadataPaths(ZooKeeper zk, |
| String logRootPath, |
| boolean ownAllocator) { |
| // Note re. persistent lock state initialization: the read lock persistent state (path) is |
| // initialized here but only used in the read handler. The reason is its more convenient and |
| // less error prone to manage all stream structure in one place. |
| final String logRootParentPath = Utils.getParent(logRootPath); |
| final String logSegmentsPath = logRootPath + LOGSEGMENTS_PATH; |
| final String maxTxIdPath = logRootPath + MAX_TXID_PATH; |
| final String lockPath = logRootPath + LOCK_PATH; |
| final String readLockPath = logRootPath + READ_LOCK_PATH; |
| final String versionPath = logRootPath + VERSION_PATH; |
| final String allocationPath = logRootPath + ALLOCATION_PATH; |
| |
| int numPaths = ownAllocator ? MetadataIndex.ALLOCATION + 1 : MetadataIndex.LOGSEGMENTS + 1; |
| List<CompletableFuture<Versioned<byte[]>>> checkFutures = Lists.newArrayListWithExpectedSize(numPaths); |
| checkFutures.add(Utils.zkGetData(zk, logRootParentPath, false)); |
| checkFutures.add(Utils.zkGetData(zk, logRootPath, false)); |
| checkFutures.add(Utils.zkGetData(zk, maxTxIdPath, false)); |
| checkFutures.add(Utils.zkGetData(zk, versionPath, false)); |
| checkFutures.add(Utils.zkGetData(zk, lockPath, false)); |
| checkFutures.add(Utils.zkGetData(zk, readLockPath, false)); |
| checkFutures.add(Utils.zkGetData(zk, logSegmentsPath, false)); |
| if (ownAllocator) { |
| checkFutures.add(Utils.zkGetData(zk, allocationPath, false)); |
| } |
| |
| return FutureUtils.collect(checkFutures); |
| } |
| |
| static boolean pathExists(Versioned<byte[]> metadata) { |
| return null != metadata.getValue() && null != metadata.getVersion(); |
| } |
| |
| static void ensureMetadataExist(Versioned<byte[]> metadata) { |
| checkNotNull(metadata.getValue()); |
| checkNotNull(metadata.getVersion()); |
| } |
| |
| static void createMissingMetadata(final ZooKeeper zk, |
| final String basePath, |
| final String logRootPath, |
| final List<Versioned<byte[]>> metadatas, |
| final List<ACL> acl, |
| final boolean ownAllocator, |
| final boolean createIfNotExists, |
| final CompletableFuture<List<Versioned<byte[]>>> promise) { |
| final List<byte[]> pathsToCreate = Lists.newArrayListWithExpectedSize(metadatas.size()); |
| final List<Op> zkOps = Lists.newArrayListWithExpectedSize(metadatas.size()); |
| CreateMode createMode = CreateMode.PERSISTENT; |
| |
| // log root parent path |
| String logRootParentPath = Utils.getParent(logRootPath); |
| if (pathExists(metadatas.get(MetadataIndex.LOG_ROOT_PARENT))) { |
| pathsToCreate.add(null); |
| } else { |
| pathsToCreate.add(EMPTY_BYTES); |
| zkOps.add(Op.create(logRootParentPath, EMPTY_BYTES, acl, createMode)); |
| } |
| |
| // log root path |
| if (pathExists(metadatas.get(MetadataIndex.LOG_ROOT))) { |
| pathsToCreate.add(null); |
| } else { |
| pathsToCreate.add(EMPTY_BYTES); |
| zkOps.add(Op.create(logRootPath, EMPTY_BYTES, acl, createMode)); |
| } |
| |
| // max id |
| if (pathExists(metadatas.get(MetadataIndex.MAX_TXID))) { |
| pathsToCreate.add(null); |
| } else { |
| byte[] zeroTxnIdData = DLUtils.serializeTransactionId(0L); |
| pathsToCreate.add(zeroTxnIdData); |
| zkOps.add(Op.create(logRootPath + MAX_TXID_PATH, zeroTxnIdData, acl, createMode)); |
| } |
| // version |
| if (pathExists(metadatas.get(MetadataIndex.VERSION))) { |
| pathsToCreate.add(null); |
| } else { |
| byte[] versionData = intToBytes(LAYOUT_VERSION); |
| pathsToCreate.add(versionData); |
| zkOps.add(Op.create(logRootPath + VERSION_PATH, versionData, acl, createMode)); |
| } |
| // lock path |
| if (pathExists(metadatas.get(MetadataIndex.LOCK))) { |
| pathsToCreate.add(null); |
| } else { |
| pathsToCreate.add(EMPTY_BYTES); |
| zkOps.add(Op.create(logRootPath + LOCK_PATH, EMPTY_BYTES, acl, createMode)); |
| } |
| // read lock path |
| if (pathExists(metadatas.get(MetadataIndex.READ_LOCK))) { |
| pathsToCreate.add(null); |
| } else { |
| pathsToCreate.add(EMPTY_BYTES); |
| zkOps.add(Op.create(logRootPath + READ_LOCK_PATH, EMPTY_BYTES, acl, createMode)); |
| } |
| // log segments path |
| if (pathExists(metadatas.get(MetadataIndex.LOGSEGMENTS))) { |
| pathsToCreate.add(null); |
| } else { |
| byte[] logSegmentsData = DLUtils.serializeLogSegmentSequenceNumber( |
| DistributedLogConstants.UNASSIGNED_LOGSEGMENT_SEQNO); |
| pathsToCreate.add(logSegmentsData); |
| zkOps.add(Op.create(logRootPath + LOGSEGMENTS_PATH, logSegmentsData, acl, createMode)); |
| } |
| // allocation path |
| if (ownAllocator) { |
| if (pathExists(metadatas.get(MetadataIndex.ALLOCATION))) { |
| pathsToCreate.add(null); |
| } else { |
| pathsToCreate.add(EMPTY_BYTES); |
| zkOps.add(Op.create(logRootPath + ALLOCATION_PATH, |
| EMPTY_BYTES, acl, createMode)); |
| } |
| } |
| if (zkOps.isEmpty()) { |
| // nothing missed |
| promise.complete(metadatas); |
| return; |
| } |
| if (!createIfNotExists) { |
| promise.completeExceptionally(new LogNotFoundException("Log " + logRootPath + " not found")); |
| return; |
| } |
| |
| getMissingPaths(zk, basePath, Utils.getParent(logRootParentPath)) |
| .whenComplete(new FutureEventListener<List<String>>() { |
| @Override |
| public void onSuccess(List<String> paths) { |
| for (String path : paths) { |
| pathsToCreate.add(EMPTY_BYTES); |
| zkOps.add( |
| 0, Op.create(path, EMPTY_BYTES, acl, createMode)); |
| } |
| executeCreateMissingPathTxn( |
| zk, |
| zkOps, |
| pathsToCreate, |
| metadatas, |
| logRootPath, |
| promise |
| ); |
| } |
| |
| @Override |
| public void onFailure(Throwable cause) { |
| promise.completeExceptionally(cause); |
| return; |
| } |
| }); |
| |
| } |
| |
| private static void executeCreateMissingPathTxn(ZooKeeper zk, |
| List<Op> zkOps, |
| List<byte[]> pathsToCreate, |
| List<Versioned<byte[]>> metadatas, |
| String logRootPath, |
| CompletableFuture<List<Versioned<byte[]>>> promise) { |
| |
| zk.multi(zkOps, new AsyncCallback.MultiCallback() { |
| @Override |
| public void processResult(int rc, String path, Object ctx, List<OpResult> resultList) { |
| if (KeeperException.Code.OK.intValue() == rc) { |
| List<Versioned<byte[]>> finalMetadatas = |
| Lists.newArrayListWithExpectedSize(metadatas.size()); |
| for (int i = 0; i < pathsToCreate.size(); i++) { |
| byte[] dataCreated = pathsToCreate.get(i); |
| if (null == dataCreated) { |
| finalMetadatas.add(metadatas.get(i)); |
| } else { |
| finalMetadatas.add(new Versioned<byte[]>(dataCreated, new LongVersion(0))); |
| } |
| } |
| promise.complete(finalMetadatas); |
| } else if (KeeperException.Code.NODEEXISTS.intValue() == rc) { |
| promise.completeExceptionally(new LogExistsException("Someone just created log " |
| + logRootPath)); |
| } else { |
| if (LOG.isDebugEnabled()) { |
| StringBuilder builder = new StringBuilder(); |
| for (OpResult result : resultList) { |
| if (result instanceof OpResult.ErrorResult) { |
| OpResult.ErrorResult errorResult = (OpResult.ErrorResult) result; |
| builder.append(errorResult.getErr()).append(","); |
| } else { |
| builder.append(0).append(","); |
| } |
| } |
| String resultCodeList = builder.substring(0, builder.length() - 1); |
| LOG.debug("Failed to create log, full rc list = {}", resultCodeList); |
| } |
| |
| promise.completeExceptionally(new ZKException("Failed to create log " + logRootPath, |
| KeeperException.Code.get(rc))); |
| } |
| } |
| }, null); |
| } |
| |
| static LogMetadataForWriter processLogMetadatas(URI uri, |
| String logName, |
| String logIdentifier, |
| List<Versioned<byte[]>> metadatas, |
| boolean ownAllocator) |
| throws UnexpectedException { |
| try { |
| // max id |
| Versioned<byte[]> maxTxnIdData = metadatas.get(MetadataIndex.MAX_TXID); |
| ensureMetadataExist(maxTxnIdData); |
| // version |
| Versioned<byte[]> versionData = metadatas.get(MetadataIndex.VERSION); |
| ensureMetadataExist(maxTxnIdData); |
| checkArgument(LAYOUT_VERSION == bytesToInt(versionData.getValue())); |
| // lock path |
| ensureMetadataExist(metadatas.get(MetadataIndex.LOCK)); |
| // read lock path |
| ensureMetadataExist(metadatas.get(MetadataIndex.READ_LOCK)); |
| // max lssn |
| Versioned<byte[]> maxLSSNData = metadatas.get(MetadataIndex.LOGSEGMENTS); |
| ensureMetadataExist(maxLSSNData); |
| try { |
| DLUtils.deserializeLogSegmentSequenceNumber(maxLSSNData.getValue()); |
| } catch (NumberFormatException nfe) { |
| throw new UnexpectedException("Invalid max sequence number found in log " + logName, nfe); |
| } |
| // allocation path |
| Versioned<byte[]> allocationData; |
| if (ownAllocator) { |
| allocationData = metadatas.get(MetadataIndex.ALLOCATION); |
| ensureMetadataExist(allocationData); |
| } else { |
| allocationData = new Versioned<byte[]>(null, null); |
| } |
| return new LogMetadataForWriter(uri, logName, logIdentifier, |
| maxLSSNData, maxTxnIdData, allocationData); |
| } catch (IllegalArgumentException iae) { |
| throw new UnexpectedException("Invalid log " + logName, iae); |
| } catch (NullPointerException npe) { |
| throw new UnexpectedException("Invalid log " + logName, npe); |
| } |
| } |
| |
| static CompletableFuture<LogMetadataForWriter> getLog(final URI uri, |
| final String logName, |
| final String logIdentifier, |
| final ZooKeeperClient zooKeeperClient, |
| final boolean ownAllocator, |
| final boolean createIfNotExists) { |
| final String logRootPath = LogMetadata.getLogRootPath(uri, logName, logIdentifier); |
| try { |
| PathUtils.validatePath(logRootPath); |
| } catch (IllegalArgumentException e) { |
| LOG.error("Illegal path value {} for stream {}", new Object[]{logRootPath, logName, e}); |
| return FutureUtils.exception(new InvalidStreamNameException(logName, "Log name is invalid")); |
| } |
| |
| try { |
| final ZooKeeper zk = zooKeeperClient.get(); |
| return checkLogMetadataPaths(zk, logRootPath, ownAllocator) |
| .thenCompose(metadatas -> { |
| CompletableFuture<List<Versioned<byte[]>>> promise = |
| new CompletableFuture<List<Versioned<byte[]>>>(); |
| createMissingMetadata( |
| zk, |
| uri.getPath(), |
| logRootPath, |
| metadatas, |
| zooKeeperClient.getDefaultACL(), |
| ownAllocator, |
| createIfNotExists, |
| promise); |
| return promise; |
| }).thenCompose(metadatas -> { |
| try { |
| return FutureUtils.value( |
| processLogMetadatas( |
| uri, |
| logName, |
| logIdentifier, |
| metadatas, |
| ownAllocator)); |
| } catch (UnexpectedException e) { |
| return FutureUtils.exception(e); |
| } |
| }); |
| } catch (ZooKeeperClient.ZooKeeperConnectionException e) { |
| return FutureUtils.exception( |
| new ZKException("Encountered zookeeper connection issue on creating log " |
| + logName, KeeperException.Code.CONNECTIONLOSS)); |
| } catch (InterruptedException e) { |
| return FutureUtils.exception(new DLInterruptedException("Interrupted on creating log " + logName, e)); |
| } |
| } |
| |
| @Override |
| public CompletableFuture<LogMetadataForWriter> getLog(final URI uri, |
| final String logName, |
| final boolean ownAllocator, |
| final boolean createIfNotExists) { |
| return getLog( |
| uri, |
| logName, |
| conf.getUnpartitionedStreamName(), |
| zooKeeperClient, |
| ownAllocator, |
| createIfNotExists); |
| } |
| |
| // |
| // Delete Log |
| // |
| |
| @Override |
| public CompletableFuture<Void> deleteLog(URI uri, final String logName) { |
| final CompletableFuture<Void> promise = new CompletableFuture<Void>(); |
| try { |
| String streamPath = LogMetadata.getLogStreamPath(uri, logName); |
| ZKUtil.deleteRecursive(zooKeeperClient.get(), streamPath, new AsyncCallback.VoidCallback() { |
| @Override |
| public void processResult(int rc, String path, Object ctx) { |
| if (KeeperException.Code.OK.intValue() != rc) { |
| FutureUtils.completeExceptionally(promise, |
| new ZKException("Encountered zookeeper issue on deleting log stream " |
| + logName, KeeperException.Code.get(rc))); |
| return; |
| } |
| FutureUtils.complete(promise, null); |
| } |
| }, null); |
| } catch (ZooKeeperClient.ZooKeeperConnectionException e) { |
| FutureUtils.completeExceptionally(promise, |
| new ZKException("Encountered zookeeper issue on deleting log stream " |
| + logName, KeeperException.Code.CONNECTIONLOSS)); |
| } catch (InterruptedException e) { |
| FutureUtils.completeExceptionally(promise, |
| new DLInterruptedException("Interrupted while deleting log stream " + logName)); |
| } catch (KeeperException e) { |
| FutureUtils.completeExceptionally(promise, |
| new ZKException("Encountered zookeeper issue on deleting log stream " |
| + logName, e)); |
| } |
| return promise; |
| } |
| |
| // |
| // Rename Log |
| // |
| |
| @Override |
| public CompletableFuture<Void> renameLog(URI uri, String oldStreamName, String newStreamName) { |
| return getLog( |
| uri, |
| oldStreamName, |
| true, |
| false |
| ).thenCompose(metadata -> renameLogMetadata(uri, metadata, newStreamName)); |
| } |
| |
| private CompletableFuture<Void> renameLogMetadata(URI uri, |
| LogMetadataForWriter oldMetadata, |
| String newStreamName) { |
| |
| |
| final LinkedList<Op> createOps = Lists.newLinkedList(); |
| final LinkedList<Op> deleteOps = Lists.newLinkedList(); |
| |
| List<ACL> acls = zooKeeperClient.getDefaultACL(); |
| |
| // get the root path |
| String oldRootPath = oldMetadata.getLogRootPath(); |
| String newRootPath = LogMetadata.getLogRootPath( |
| uri, newStreamName, conf.getUnpartitionedStreamName()); |
| |
| // 0. the log path |
| deleteOps.addFirst(Op.delete( |
| LogMetadata.getLogStreamPath(uri, oldMetadata.getLogName()), -1)); |
| |
| // 1. the root path |
| createOps.addLast(Op.create( |
| newRootPath, EMPTY_BYTES, acls, CreateMode.PERSISTENT)); |
| deleteOps.addFirst(Op.delete( |
| oldRootPath, -1)); |
| |
| // 2. max id |
| Versioned<byte[]> maxTxIdData = oldMetadata.getMaxTxIdData(); |
| deleteOldPathAndCreateNewPath( |
| oldRootPath, MAX_TXID_PATH, maxTxIdData, |
| newRootPath, DLUtils.serializeTransactionId(0L), acls, |
| createOps, deleteOps |
| ); |
| |
| // 3. version |
| createOps.addLast(Op.create( |
| newRootPath + VERSION_PATH, intToBytes(LAYOUT_VERSION), acls, CreateMode.PERSISTENT)); |
| deleteOps.addFirst(Op.delete( |
| oldRootPath + VERSION_PATH, -1)); |
| |
| // 4. lock path (NOTE: if the stream is locked by a writer, then the delete will fail as you can not |
| // delete the lock path if children is not empty. |
| createOps.addLast(Op.create( |
| newRootPath + LOCK_PATH, EMPTY_BYTES, acls, CreateMode.PERSISTENT)); |
| deleteOps.addFirst(Op.delete( |
| oldRootPath + LOCK_PATH, -1)); |
| |
| // 5. read lock path (NOTE: same reason as the write lock) |
| createOps.addLast(Op.create( |
| newRootPath + READ_LOCK_PATH, EMPTY_BYTES, acls, CreateMode.PERSISTENT)); |
| deleteOps.addFirst(Op.delete( |
| oldRootPath + READ_LOCK_PATH, -1)); |
| |
| // 6. allocation path |
| Versioned<byte[]> allocationData = oldMetadata.getAllocationData(); |
| deleteOldPathAndCreateNewPath( |
| oldRootPath, ALLOCATION_PATH, allocationData, |
| newRootPath, EMPTY_BYTES, acls, |
| createOps, deleteOps); |
| |
| // 7. log segments |
| Versioned<byte[]> maxLSSNData = oldMetadata.getMaxLSSNData(); |
| deleteOldPathAndCreateNewPath( |
| oldRootPath, LOGSEGMENTS_PATH, maxLSSNData, |
| newRootPath, DLUtils.serializeLogSegmentSequenceNumber(UNASSIGNED_LOGSEGMENT_SEQNO), acls, |
| createOps, deleteOps); |
| |
| // 8. copy the log segments |
| CompletableFuture<List<LogSegmentMetadata>> segmentsFuture; |
| if (pathExists(maxLSSNData)) { |
| segmentsFuture = getLogSegments(zooKeeperClient, oldRootPath + LOGSEGMENTS_PATH); |
| } else { |
| segmentsFuture = FutureUtils.value(Collections.emptyList()); |
| } |
| return segmentsFuture |
| // copy the segments |
| .thenApply(segments -> { |
| for (LogSegmentMetadata segment : segments) { |
| deleteOldSegmentAndCreateNewSegment( |
| segment, |
| newRootPath + LOGSEGMENTS_PATH, |
| acls, |
| createOps, |
| deleteOps); |
| } |
| return null; |
| }) |
| // get the missing paths |
| .thenCompose(ignored -> |
| getMissingPaths(zooKeeperClient, uri, newStreamName) |
| ) |
| // create the missing paths and execute the rename transaction |
| .thenCompose(paths -> { |
| for (String path : paths) { |
| createOps.addFirst(Op.create( |
| path, EMPTY_BYTES, acls, CreateMode.PERSISTENT)); |
| } |
| return executeRenameTxn(oldRootPath, newRootPath, createOps, deleteOps); |
| }); |
| } |
| |
| @VisibleForTesting |
| static CompletableFuture<List<String>> getMissingPaths(ZooKeeperClient zkc, URI uri, String logName) { |
| ZooKeeper zk; |
| try { |
| zk = zkc.get(); |
| } catch (ZooKeeperConnectionException | InterruptedException e) { |
| return FutureUtils.exception(e); |
| } |
| String basePath = uri.getPath(); |
| String logStreamPath = LogMetadata.getLogStreamPath(uri, logName); |
| return getMissingPaths(zk, basePath, logStreamPath); |
| } |
| |
| @VisibleForTesting |
| static CompletableFuture<List<String>> getMissingPaths(ZooKeeper zk, String basePath, String logStreamPath) { |
| LinkedList<String> missingPaths = Lists.newLinkedList(); |
| CompletableFuture<List<String>> future = FutureUtils.createFuture(); |
| existPath(zk, logStreamPath, basePath, missingPaths, future); |
| return future; |
| } |
| |
| private static void existPath(ZooKeeper zk, |
| String path, |
| String basePath, |
| LinkedList<String> missingPaths, |
| CompletableFuture<List<String>> future) { |
| if (basePath.equals(path)) { |
| future.complete(missingPaths); |
| return; |
| } |
| zk.exists(path, false, (rc, path1, ctx, stat) -> { |
| if (Code.OK.intValue() != rc && Code.NONODE.intValue() != rc) { |
| future.completeExceptionally(new ZKException("Failed to check existence of path " + path1, |
| Code.get(rc))); |
| return; |
| } |
| |
| if (Code.OK.intValue() == rc) { |
| future.complete(missingPaths); |
| return; |
| } |
| |
| missingPaths.addLast(path); |
| String parentPath = Utils.getParent(path); |
| existPath(zk, parentPath, basePath, missingPaths, future); |
| }, null); |
| } |
| |
| private CompletableFuture<Void> executeRenameTxn(String oldLogPath, |
| String newLogPath, |
| LinkedList<Op> createOps, |
| LinkedList<Op> deleteOps) { |
| CompletableFuture<Void> future = FutureUtils.createFuture(); |
| List<Op> zkOps = Lists.newArrayListWithExpectedSize(createOps.size() + deleteOps.size()); |
| zkOps.addAll(createOps); |
| zkOps.addAll(deleteOps); |
| |
| if (LOG.isDebugEnabled()) { |
| for (Op op : zkOps) { |
| if (op instanceof Create) { |
| Create create = (Create) op; |
| LOG.debug("op : create {}", create.getPath()); |
| } else if (op instanceof Delete) { |
| Delete delete = (Delete) op; |
| LOG.debug("op : delete {}, record = {}", delete.getPath(), op.toRequestRecord()); |
| } else { |
| LOG.debug("op : {}", op); |
| } |
| } |
| } |
| |
| try { |
| zooKeeperClient.get().multi(zkOps, (rc, path, ctx, opResults) -> { |
| if (Code.OK.intValue() == rc) { |
| future.complete(null); |
| } else if (Code.NODEEXISTS.intValue() == rc) { |
| future.completeExceptionally(new LogExistsException("Someone just created new log " + newLogPath)); |
| } else if (Code.NOTEMPTY.intValue() == rc) { |
| future.completeExceptionally(new LockingException(oldLogPath + LOCK_PATH, |
| "Someone is holding a lock on log " + oldLogPath)); |
| } else { |
| future.completeExceptionally(new ZKException("Failed to rename log " |
| + oldLogPath + " to " + newLogPath + " at path " + path, Code.get(rc))); |
| } |
| }, null); |
| } catch (ZooKeeperConnectionException e) { |
| future.completeExceptionally(e); |
| } catch (InterruptedException e) { |
| future.completeExceptionally(e); |
| } |
| return future; |
| } |
| |
| private static void deleteOldSegmentAndCreateNewSegment(LogSegmentMetadata oldMetadata, |
| String newSegmentsPath, |
| List<ACL> acls, |
| LinkedList<Op> createOps, |
| LinkedList<Op> deleteOps) { |
| createOps.addLast(Op.create( |
| newSegmentsPath + "/" + oldMetadata.getZNodeName(), |
| oldMetadata.getFinalisedData().getBytes(UTF_8), |
| acls, |
| CreateMode.PERSISTENT)); |
| deleteOps.addFirst(Op.delete( |
| oldMetadata.getZkPath(), |
| -1)); |
| } |
| |
| private static void deleteOldPathAndCreateNewPath(String oldRootPath, |
| String nodePath, |
| Versioned<byte[]> pathData, |
| String newRootPath, |
| byte[] initData, |
| List<ACL> acls, |
| LinkedList<Op> createOps, |
| LinkedList<Op> deleteOps) { |
| if (pathExists(pathData)) { |
| createOps.addLast(Op.create( |
| newRootPath + nodePath, pathData.getValue(), acls, CreateMode.PERSISTENT)); |
| deleteOps.addFirst(Op.delete( |
| oldRootPath + nodePath, (int) ((LongVersion) pathData.getVersion()).getLongVersion())); |
| } else { |
| createOps.addLast(Op.create( |
| newRootPath + nodePath, initData, acls, CreateMode.PERSISTENT)); |
| } |
| } |
| |
| @VisibleForTesting |
| static CompletableFuture<List<LogSegmentMetadata>> getLogSegments(ZooKeeperClient zk, |
| String logSegmentsPath) { |
| CompletableFuture<List<LogSegmentMetadata>> future = FutureUtils.createFuture(); |
| try { |
| zk.get().getChildren(logSegmentsPath, false, (rc, path, ctx, children, stat) -> { |
| if (Code.OK.intValue() != rc) { |
| if (Code.NONODE.intValue() == rc) { |
| future.completeExceptionally(new LogNotFoundException("Log " + path + " not found")); |
| } else { |
| future.completeExceptionally(new ZKException("Failed to get log segments from " + path, |
| Code.get(rc))); |
| } |
| return; |
| } |
| |
| // get all the segments |
| List<CompletableFuture<LogSegmentMetadata>> futures = |
| Lists.newArrayListWithExpectedSize(children.size()); |
| for (String child : children) { |
| futures.add(LogSegmentMetadata.read(zk, logSegmentsPath + "/" + child)); |
| } |
| FutureUtils.proxyTo( |
| FutureUtils.collect(futures), |
| future); |
| }, null); |
| } catch (ZooKeeperConnectionException e) { |
| future.completeExceptionally(e); |
| } catch (InterruptedException e) { |
| future.completeExceptionally(e); |
| } |
| return future; |
| } |
| |
| } |
