Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,9 @@ public class ClientConfiguration extends AbstractConfiguration<ClientConfigurati
protected static final String ENABLE_BOOKIE_FAILURE_TRACKING = "enableBookieFailureTracking";
protected static final String BOOKIE_FAILURE_HISTORY_EXPIRATION_MS = "bookieFailureHistoryExpirationMSec";

// Discovery
protected static final String FOLLOW_BOOKIE_ADDRESS_TRACKING = "enableBookieAddressTracking";

// Names of dynamic features
protected static final String DISABLE_ENSEMBLE_CHANGE_FEATURE_NAME = "disableEnsembleChangeFeatureName";

Expand Down Expand Up @@ -1764,6 +1767,27 @@ public ClientConfiguration setDelayEnsembleChange(boolean enabled) {
return this;
}

/**
* Whether to enable bookie address changes tracking.
*
* @return flag to enable/disable bookie address changes tracking
*/
public boolean getEnableBookieAddressTracking() {
return getBoolean(FOLLOW_BOOKIE_ADDRESS_TRACKING, true);
}

/**
* Enable/Disable bookie address changes tracking.
*
* @param value
* flag to enable/disable bookie address changes tracking
* @return client configuration.
*/
public ClientConfiguration setEnableBookieAddressTracking(boolean value) {
setProperty(FOLLOW_BOOKIE_ADDRESS_TRACKING, value);
return this;
}

/**
* Whether to enable bookie failure tracking.
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -182,18 +182,26 @@ public void close() {
private WatchTask watchReadOnlyBookiesTask = null;
private final ConcurrentHashMap<BookieId, Versioned<BookieServiceInfo>> bookieServiceInfoCache =
new ConcurrentHashMap<>();

private final Watcher bookieServiceInfoCacheInvalidation;
private final boolean bookieAddressTracking;
// registration paths
private final String bookieRegistrationPath;
private final String bookieAllRegistrationPath;
private final String bookieReadonlyRegistrationPath;

public ZKRegistrationClient(ZooKeeper zk,
String ledgersRootPath,
ScheduledExecutorService scheduler) {
ScheduledExecutorService scheduler,
boolean bookieAddressTracking) {
this.zk = zk;
this.scheduler = scheduler;

// Following Bookie Network Address Changes is an expensive operation
// as it requires additional ZooKeeper watches
// we can disable this feature, in case the BK cluster has only
// static addresses
this.bookieAddressTracking = bookieAddressTracking;
this.bookieServiceInfoCacheInvalidation = bookieAddressTracking
? new BookieServiceInfoCacheInvalidationWatcher() : null;
this.bookieRegistrationPath = ledgersRootPath + "/" + AVAILABLE_NODE;
this.bookieAllRegistrationPath = ledgersRootPath + "/" + COOKIE_NODE;
this.bookieReadonlyRegistrationPath = this.bookieRegistrationPath + "/" + READONLY;
Expand All @@ -204,6 +212,10 @@ public void close() {
// no-op
}

public boolean isBookieAddressTracking() {
return bookieAddressTracking;
}

public ZooKeeper getZk() {
return zk;
}
Expand Down Expand Up @@ -243,12 +255,13 @@ public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(Book
* @param bookieId
* @return an handle to the result of the operation.
*/
private CompletableFuture<Versioned<BookieServiceInfo>> readBookieServiceInfo(BookieId bookieId) {
private CompletableFuture<Versioned<BookieServiceInfo>> readBookieServiceInfoAsync(BookieId bookieId) {
String pathAsWritable = bookieRegistrationPath + "/" + bookieId;
String pathAsReadonly = bookieReadonlyRegistrationPath + "/" + bookieId;

CompletableFuture<Versioned<BookieServiceInfo>> promise = new CompletableFuture<>();
zk.getData(pathAsWritable, null, (int rc, String path, Object o, byte[] bytes, Stat stat) -> {
zk.getData(pathAsWritable, bookieServiceInfoCacheInvalidation,
(int rc, String path, Object o, byte[] bytes, Stat stat) -> {
if (KeeperException.Code.OK.intValue() == rc) {
try {
BookieServiceInfo bookieServiceInfo = deserializeBookieServiceInfo(bookieId, bytes);
Expand All @@ -265,7 +278,8 @@ private CompletableFuture<Versioned<BookieServiceInfo>> readBookieServiceInfo(Bo
}
} else if (KeeperException.Code.NONODE.intValue() == rc) {
// not found, looking for a readonly bookie
zk.getData(pathAsReadonly, null, (int rc2, String path2, Object o2, byte[] bytes2, Stat stat2) -> {
zk.getData(pathAsReadonly, bookieServiceInfoCacheInvalidation,
(int rc2, String path2, Object o2, byte[] bytes2, Stat stat2) -> {
if (KeeperException.Code.OK.intValue() == rc2) {
try {
BookieServiceInfo bookieServiceInfo = deserializeBookieServiceInfo(bookieId, bytes2);
Expand Down Expand Up @@ -344,7 +358,7 @@ private CompletableFuture<Versioned<Set<BookieId>>> getChildren(String regPath,
for (BookieId id : bookies) {
// update the cache for new bookies
if (!bookieServiceInfoCache.containsKey(id)) {
bookieInfoUpdated.add(readBookieServiceInfo(id));
bookieInfoUpdated.add(readBookieServiceInfoAsync(id));
}
}
if (bookieInfoUpdated.isEmpty()) {
Expand Down Expand Up @@ -447,4 +461,46 @@ private static HashSet<BookieId> convertToBookieAddresses(List<String> children)
return newBookieAddrs;
}

private static BookieId stripBookieIdFromPath(String path) {
final int slash = path.lastIndexOf('/');
if (slash >= 0) {
try {
return BookieId.parse(path.substring(slash + 1));
} catch (IllegalArgumentException e) {
log.warn("Cannot decode bookieId from {}", path, e);
}
}
return null;
}

private class BookieServiceInfoCacheInvalidationWatcher implements Watcher {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How frequently can a bookieID change? What factors drive its change?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see these known cases:

  • a bookie with a dynamic network address
  • changing configuration (change DNS name....)
  • you change configuration and add new endpoints (this will happen in the future, when for instance @sijie will contribute the pure TLS endpoint)

So basically I think they won't change very ofter

@Override
public void process(WatchedEvent we) {
log.debug("zk event {} for {} state {}", we.getType(), we.getPath(), we.getState());
if (we.getState() == KeeperState.Expired) {
log.info("zk session expired, invalidating cache");
bookieServiceInfoCache.clear();
return;
}
BookieId bookieId = stripBookieIdFromPath(we.getPath());
if (bookieId == null) {
return;
}
switch (we.getType()) {
case NodeDeleted:
log.info("Invalidate cache for {}", bookieId);
bookieServiceInfoCache.remove(bookieId);
break;
case NodeDataChanged:
log.info("refresh cache for {}", bookieId);
readBookieServiceInfoAsync(bookieId);
break;
default:
log.debug("ignore cache event {} for {}", we.getType(), bookieId);
break;
}
}
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -515,7 +515,8 @@ public boolean nukeExistingCluster() throws Exception {
try (RegistrationClient regClient = new ZKRegistrationClient(
zk,
ledgersRootPath,
null
null,
false
)) {
if (availableNodeExists) {
Collection<BookieId> rwBookies = FutureUtils
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ public class ZKMetadataClientDriver
ClientConfiguration clientConf;
ScheduledExecutorService scheduler;
RegistrationClient regClient;
boolean bookieAddressTracking = true;

@Override
public synchronized MetadataClientDriver initialize(ClientConfiguration conf,
Expand All @@ -70,6 +71,7 @@ public synchronized MetadataClientDriver initialize(ClientConfiguration conf,
this.statsLogger = statsLogger;
this.clientConf = conf;
this.scheduler = scheduler;
this.bookieAddressTracking = conf.getEnableBookieAddressTracking();
return this;
}

Expand All @@ -79,7 +81,8 @@ public synchronized RegistrationClient getRegistrationClient() {
regClient = new ZKRegistrationClient(
zk,
ledgersRootPath,
scheduler);
scheduler,
bookieAddressTracking);
}
return regClient;
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,178 @@
/*
*
* 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.bookkeeper.client;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException.BKBookieHandleNotAvailableException;
import org.apache.bookkeeper.client.api.BookKeeper;
import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.client.api.WriteHandle;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.discover.ZKRegistrationClient;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.bookkeeper.util.PortManager;
import org.junit.Test;

/**
* Tests of the main BookKeeper client and the BP-41 bookieAddressTracking feature.
*/
@Slf4j
public class BookieNetworkAddressChangeTest extends BookKeeperClusterTestCase {

public BookieNetworkAddressChangeTest() {
super(1);
this.useUUIDasBookieId = true;
}

@Test
public void testFollowBookieAddressChange() throws Exception {
ClientConfiguration conf = new ClientConfiguration();
conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
try (BookKeeper bkc = BookKeeper.newBuilder(conf)
.build();) {
long lId;
try (WriteHandle h = bkc
.newCreateLedgerOp()
.withAckQuorumSize(1)
.withEnsembleSize(1)
.withWriteQuorumSize(1)
.withPassword(new byte[0])
.execute()
.get();) {
lId = h.getId();
h.append("foo".getBytes("utf-8"));
}

// restart bookie, change port
// on metadata we have a bookieId, not the network address
ServerConfiguration thisServerConf = new ServerConfiguration(baseConf);
thisServerConf.setBookiePort(PortManager.nextFreePort());
restartBookies(thisServerConf);

try (ReadHandle h = bkc
.newOpenLedgerOp()
.withLedgerId(lId)
.withRecovery(true)
.withPassword(new byte[0])
.execute()
.get()) {
assertEquals(0, h.getLastAddConfirmed());
try (LedgerEntries entries = h.read(0, 0);) {
assertEquals("foo", new String(entries.getEntry(0).getEntryBytes(), "utf-8"));
}
}
}
}

@Test
public void testFollowBookieAddressChangeTrckingDisabled() throws Exception {
ClientConfiguration conf = new ClientConfiguration();
conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
conf.setEnableBookieAddressTracking(false);
try (BookKeeper bkc = BookKeeper.newBuilder(conf)
.build();) {
long lId;
try (WriteHandle h = bkc
.newCreateLedgerOp()
.withAckQuorumSize(1)
.withEnsembleSize(1)
.withWriteQuorumSize(1)
.withPassword(new byte[0])
.execute()
.get();) {
lId = h.getId();
h.append("foo".getBytes("utf-8"));
}

// restart bookie, change port
// on metadata we have a bookieId, not the network address
ServerConfiguration thisServerConf = new ServerConfiguration(baseConf);
thisServerConf.setBookiePort(PortManager.nextFreePort());
restartBookies(thisServerConf);

try (ReadHandle h = bkc
.newOpenLedgerOp()
.withLedgerId(lId)
.withRecovery(true)
.withPassword(new byte[0])
.execute()
.get()) {
try (LedgerEntries entries = h.read(0, 0);) {
fail("Should not be able to connect to the bookie with Bookie Address Tracking Disabled");
} catch (BKBookieHandleNotAvailableException expected) {
}
}
}
}

@Test
public void testFollowBookieAddressChangeZkSessionExpire() throws Exception {
ClientConfiguration conf = new ClientConfiguration();
conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
try (BookKeeper bkc = BookKeeper.newBuilder(conf)
.build();) {
long lId;
try (WriteHandle h = bkc
.newCreateLedgerOp()
.withAckQuorumSize(1)
.withEnsembleSize(1)
.withWriteQuorumSize(1)
.withPassword(new byte[0])
.execute()
.get();) {
lId = h.getId();
h.append("foo".getBytes("utf-8"));
}

log.error("expiring ZK session!");
// expire zk session
ZKRegistrationClient regClient = (ZKRegistrationClient) ((org.apache.bookkeeper.client.BookKeeper) bkc)
.getMetadataClientDriver()
.getRegistrationClient();

regClient.getZk().getTestable().injectSessionExpiration();

// restart bookie, change port
// on metadata we have a bookieId, not the network address
ServerConfiguration thisServerConf = new ServerConfiguration(baseConf);
thisServerConf.setBookiePort(PortManager.nextFreePort());
restartBookies(thisServerConf);

try (ReadHandle h = bkc
.newOpenLedgerOp()
.withLedgerId(lId)
.withRecovery(true)
.withPassword(new byte[0])
.execute()
.get()) {
assertEquals(0, h.getLastAddConfirmed());
try (LedgerEntries entries = h.read(0, 0);) {
assertEquals("foo", new String(entries.getEntry(0).getEntryBytes(), "utf-8"));
}
}
}
}
}
Loading