Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
48 commits
Select commit Hold shift + click to select a range
b1e3524
BP-41 client side implementation
eolivelli Jul 25, 2020
ae70d45
continue work
Jul 27, 2020
daf0d95
other changes
Jul 28, 2020
2d6ef74
fix build
Jul 29, 2020
6e70bcf
fix build
Jul 29, 2020
e283771
fix dlog
Jul 30, 2020
597b0ee
use BookieId
Jul 30, 2020
b528887
more
Jul 30, 2020
03cb283
more
Jul 30, 2020
9366794
fix build
eolivelli Aug 4, 2020
028793a
fix checkstyle
eolivelli Aug 4, 2020
98cabb2
fix BookKeeper init
eolivelli Aug 5, 2020
4298263
fix getBookieId
eolivelli Aug 5, 2020
d587eba
fix build
eolivelli Aug 5, 2020
55efe46
more checkstyle
eolivelli Aug 7, 2020
d468a1b
more checkstyle
eolivelli Aug 7, 2020
91e8616
more checkstyle
eolivelli Aug 7, 2020
5cb9f7a
more checkstyle
eolivelli Aug 7, 2020
91af731
more checkstyle
eolivelli Aug 8, 2020
27e0c93
more checkstyle
eolivelli Aug 8, 2020
3ce268c
remove file
eolivelli Aug 8, 2020
d7382fe
add config
eolivelli Aug 10, 2020
c256f98
Fix connection to down bookie
eolivelli Aug 11, 2020
206836b
temporary fixes
eolivelli Aug 12, 2020
6ddd579
more fixes and add debug
eolivelli Aug 12, 2020
1cd1bf5
Merge remote-tracking branch 'origin/master' into bp-41-bookieid-impl
eolivelli Aug 14, 2020
5e17020
fix tests, add BookieId validation
eolivelli Aug 14, 2020
39c8936
Fix more tests
eolivelli Aug 14, 2020
6429f91
try resolve deadlock
eolivelli Aug 16, 2020
af90119
cache eagerly bookieinfo
eolivelli Aug 16, 2020
169a160
populate cache correctly
eolivelli Aug 16, 2020
8baabfa
fix fallback logic
eolivelli Aug 16, 2020
df00ab8
fix test
eolivelli Aug 16, 2020
2fc0cad
fix checkstyle
eolivelli Aug 16, 2020
9944032
drop final qualifier
eolivelli Aug 16, 2020
77019e4
rerunFailingTestsCount
eolivelli Aug 17, 2020
9dddd2b
fix more tests
eolivelli Aug 17, 2020
2acff70
fix tests
eolivelli Aug 17, 2020
d9ae2af
fix more tests
eolivelli Aug 17, 2020
92f96aa
fix test
eolivelli Aug 27, 2020
d8ce265
use legacy resolver only for dummy nodes
eolivelli Aug 27, 2020
5b51d57
fix a couple of tests
eolivelli Aug 31, 2020
dc25ba7
remove useless code
eolivelli Aug 31, 2020
0c729e2
fix checkstyle
eolivelli Aug 31, 2020
c4de07b
Rename class
eolivelli Aug 31, 2020
d244b7f
drop useless class
eolivelli Aug 31, 2020
59817ad
more clean up
eolivelli Aug 31, 2020
a3bd16b
reserve 'readonly' bookieId
eolivelli Aug 31, 2020
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 @@ -35,6 +35,7 @@
import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.net.BookieId;

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.

This is good. I am just wondering the real usecase behind this. Do you expect your bookies to change the ip and nodenames ? We had this requirement earlier when we moved to k8s where ip addresses could change, but we worked around by keeping all clients and bookies in the same k8s cluster where the node names are constant.

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.

@jvrao primary use cases are:

  • dynamic addresses for Bookies (or simply the ability to change hostname/ip for a bookie)
  • support more endpoints and protocols for Bookie (now we are stuck to one single host:port endpoint, and we had to use StartTLS, with BP-38 the Bookie advertises all of the endpoints and thus the client will be able in the future to connect to the preferred one)

see https://github.com/apache/bookkeeper/blob/master/site/bps/BP-41-bookieid.md for more details

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I recalled that, user from puslar side reported same requirements, they want to change a hostname/ip for a bookie

import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookieClient;
import org.apache.bookkeeper.proto.BookieClientImpl;
Expand Down Expand Up @@ -63,7 +64,7 @@ static class LatencyCallback implements WriteCallback {
boolean complete;
@Override
public synchronized void writeComplete(int rc, long ledgerId, long entryId,
BookieSocketAddress addr, Object ctx) {
BookieId addr, Object ctx) {
if (rc != 0) {
LOG.error("Got error " + rc);
}
Expand All @@ -85,7 +86,7 @@ static class ThroughputCallback implements WriteCallback {
int waitingCount = Integer.MAX_VALUE;
@Override
public synchronized void writeComplete(int rc, long ledgerId, long entryId,
BookieSocketAddress addr, Object ctx) {
BookieId addr, Object ctx) {
if (rc != 0) {
LOG.error("Got error " + rc);
}
Expand Down Expand Up @@ -178,7 +179,7 @@ public static void main(String[] args)

ClientConfiguration conf = new ClientConfiguration();
BookieClient bc = new BookieClientImpl(conf, eventLoop, PooledByteBufAllocator.DEFAULT, executor, scheduler,
NullStatsLogger.INSTANCE);
NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
LatencyCallback lc = new LatencyCallback();

ThroughputCallback tc = new ThroughputCallback();
Expand All @@ -191,7 +192,7 @@ public static void main(String[] args)
toSend.writeLong(ledger);
toSend.writeLong(entry);
toSend.writerIndex(toSend.capacity());
bc.addEntry(new BookieSocketAddress(addr, port), ledger, new byte[20],
bc.addEntry(new BookieSocketAddress(addr, port).toBookieId(), ledger, new byte[20],
entry, ByteBufList.get(toSend), tc, null, BookieProtocol.FLAG_NONE,
false, WriteFlag.NONE);
}
Expand All @@ -209,7 +210,7 @@ public static void main(String[] args)
toSend.writeLong(entry);
toSend.writerIndex(toSend.capacity());
lc.resetComplete();
bc.addEntry(new BookieSocketAddress(addr, port), ledger, new byte[20],
bc.addEntry(new BookieSocketAddress(addr, port).toBookieId(), ledger, new byte[20],
entry, ByteBufList.get(toSend), lc, null,
BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
lc.waitForComplete();
Expand All @@ -228,7 +229,7 @@ public static void main(String[] args)
toSend.writeLong(ledger);
toSend.writeLong(entry);
toSend.writerIndex(toSend.capacity());
bc.addEntry(new BookieSocketAddress(addr, port), ledger, new byte[20],
bc.addEntry(new BookieSocketAddress(addr, port).toBookieId(), ledger, new byte[20],
entry, ByteBufList.get(toSend), tc, null, BookieProtocol.FLAG_NONE,
false, WriteFlag.NONE);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ public void testThroughputLatency() throws Exception {

@Test
public void testBookie() throws Exception {
BookieSocketAddress bookie = getBookie(0);
BookieSocketAddress bookie = getBookieAddress(0);
BenchBookie.main(new String[] {
"--host", bookie.getSocketAddress().getHostName(),
"--port", String.valueOf(bookie.getPort()),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,15 +30,15 @@ public static String toJson(Object object) throws ParseJsonException {
try {
return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(object);
} catch (Exception e) {
throw new ParseJsonException("Failed to serialize Object to Json string");
throw new ParseJsonException("Failed to serialize Object to Json string", e);
}
}

public static <T> T fromJson(String jsonStr, Class<T> valueType) throws ParseJsonException {
try {
return mapper.readValue(jsonStr, valueType);
} catch (Exception e) {
throw new ParseJsonException("Failed to deserialize Object from Json string");
throw new ParseJsonException("Failed to deserialize Object from Json string", e);
}
}

Expand All @@ -49,5 +49,9 @@ public static class ParseJsonException extends Exception {
public ParseJsonException(String message) {
super(message);
}

public ParseJsonException(String message, Throwable cause) {
super(message, cause);
}
}
}
1 change: 1 addition & 0 deletions bookkeeper-server/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,7 @@
<artifactId>maven-surefire-plugin</artifactId>
<version>${maven-surefire-plugin.version}</version>
<configuration>
<trimStackTrace>false</trimStackTrace>
<properties>
<property>
<name>listener</name>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@
import org.apache.bookkeeper.meta.MetadataBookieDriver;
import org.apache.bookkeeper.meta.MetadataDrivers;
import org.apache.bookkeeper.meta.exceptions.MetadataException;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.net.DNS;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
Expand Down Expand Up @@ -188,7 +189,7 @@ public long getEntry() {
static class NopWriteCallback implements WriteCallback {
@Override
public void writeComplete(int rc, long ledgerId, long entryId,
BookieSocketAddress addr, Object ctx) {
BookieId addr, Object ctx) {
if (LOG.isDebugEnabled()) {
LOG.debug("Finished writing entry {} @ ledger {} for {} : {}",
entryId, ledgerId, addr, rc);
Expand Down Expand Up @@ -302,11 +303,11 @@ private void checkIfDirsOnSameDiskPartition(List<File> dirs) throws DiskPartitio
}
}

static List<BookieSocketAddress> possibleBookieIds(ServerConfiguration conf)
static List<BookieId> possibleBookieIds(ServerConfiguration conf)
throws BookieException {
// we need to loop through all possible bookie identifiers to ensure it is treated as a new environment
// just because of bad configuration
List<BookieSocketAddress> addresses = Lists.newArrayListWithExpectedSize(3);
List<BookieId> addresses = Lists.newArrayListWithExpectedSize(3);
// we are checking all possibilities here, so we don't need to fail if we can only get
// loopback address. it will fail anyway when the bookie attempts to listen on loopback address.
try {
Expand All @@ -316,17 +317,17 @@ static List<BookieSocketAddress> possibleBookieIds(ServerConfiguration conf)
.setUseHostNameAsBookieID(false)
.setAdvertisedAddress(null)
.setAllowLoopback(true)
));
).toBookieId());
// host name
addresses.add(getBookieAddress(
new ServerConfiguration(conf)
.setUseHostNameAsBookieID(true)
.setAdvertisedAddress(null)
.setAllowLoopback(true)
));
).toBookieId());
// advertised address
if (null != conf.getAdvertisedAddress()) {
addresses.add(getBookieAddress(conf));
addresses.add(getBookieAddress(conf).toBookieId());
}
} catch (UnknownHostException e) {
throw new UnknownBookieIdException(e);
Expand All @@ -336,10 +337,10 @@ static List<BookieSocketAddress> possibleBookieIds(ServerConfiguration conf)

static Versioned<Cookie> readAndVerifyCookieFromRegistrationManager(
Cookie masterCookie, RegistrationManager rm,
List<BookieSocketAddress> addresses, boolean allowExpansion)
List<BookieId> addresses, boolean allowExpansion)
throws BookieException {
Versioned<Cookie> rmCookie = null;
for (BookieSocketAddress address : addresses) {
for (BookieId address : addresses) {
try {
rmCookie = Cookie.readFromRegistrationManager(rm, address);
// If allowStorageExpansion option is set, we should
Expand Down Expand Up @@ -419,7 +420,7 @@ public static void checkEnvironmentWithStorageExpansion(
// 3. read the cookie from registration manager. it is the `source-of-truth` of a given bookie.
// if it doesn't exist in registration manager, this bookie is a new bookie, otherwise it is
// an old bookie.
List<BookieSocketAddress> possibleBookieIds = possibleBookieIds(conf);
List<BookieId> possibleBookieIds = possibleBookieIds(conf);
final Versioned<Cookie> rmCookie = readAndVerifyCookieFromRegistrationManager(
masterCookie, rm, possibleBookieIds, allowExpansion);

Expand Down Expand Up @@ -532,6 +533,10 @@ private static void verifyDirsForStorageExpansion(
}
}

public static BookieId getBookieId(ServerConfiguration conf) throws UnknownHostException {
return getBookieAddress(conf).toBookieId();
}

/**
* Return the configured address of the bookie.
*/
Expand Down Expand Up @@ -1485,7 +1490,7 @@ static class CounterCallback implements WriteCallback {
int count;

@Override
public synchronized void writeComplete(int rc, long l, long e, BookieSocketAddress addr, Object ctx) {
public synchronized void writeComplete(int rc, long l, long e, BookieId addr, Object ctx) {
count--;
if (count == 0) {
notifyAll();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@
import org.apache.bookkeeper.client.api.LedgerMetadata;
import org.apache.bookkeeper.common.annotation.InterfaceAudience.Private;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.tools.cli.commands.autorecovery.ListUnderReplicatedCommand;
import org.apache.bookkeeper.tools.cli.commands.autorecovery.LostBookieRecoveryDelayCommand;
import org.apache.bookkeeper.tools.cli.commands.autorecovery.ToggleCommand;
Expand Down Expand Up @@ -600,7 +600,7 @@ int runCmd(CommandLine cmdLine) throws Exception {

boolean printMsg = cmdLine.hasOption("m");
boolean forceRecovery = cmdLine.hasOption("r");
final BookieSocketAddress bookie;
final BookieId bookie;
String bookieAddress;
if (cmdLine.hasOption("b")) {
// A particular bookie was specified
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.MetadataBookieDriver;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.stats.Gauge;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
Expand Down Expand Up @@ -79,7 +80,7 @@ public class BookieStateManager implements StateManager {
private final AtomicBoolean forceReadOnly = new AtomicBoolean(false);
private volatile boolean availableForHighPriorityWrites = true;

private final String bookieId;
private final BookieId bookieId;
private ShutdownHandler shutdownHandler;
private final Supplier<RegistrationManager> rm;
// Expose Stats
Expand All @@ -101,7 +102,7 @@ public BookieStateManager(ServerConfiguration conf,
ledgerDirsManager.getAllLedgerDirs(),
() -> {
try {
return Bookie.getBookieAddress(conf).toString();
return Bookie.getBookieId(conf);
} catch (UnknownHostException e) {
throw new UncheckedIOException("Failed to resolve bookie id", e);
}
Expand All @@ -112,7 +113,7 @@ public BookieStateManager(ServerConfiguration conf,
StatsLogger statsLogger,
Supplier<RegistrationManager> rm,
List<File> statusDirs,
Supplier<String> bookieIdSupplier,
Supplier<BookieId> bookieIdSupplier,
Supplier<BookieServiceInfo> bookieServiceInfoProvider) throws IOException {
this.conf = conf;
this.rm = rm;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@
import org.apache.bookkeeper.bookie.BookieException.UnknownBookieIdException;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.proto.DataFormats.CookieFormat;
import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.bookkeeper.versioning.LongVersion;
Expand Down Expand Up @@ -234,14 +234,14 @@ public void writeToFile (File versionFile) throws IOException {
*/
public void writeToRegistrationManager(RegistrationManager rm, ServerConfiguration conf, Version version)
throws BookieException {
BookieSocketAddress address = null;
BookieId address = null;
try {
address = Bookie.getBookieAddress(conf);
address = Bookie.getBookieId(conf);
} catch (UnknownHostException e) {
throw new UnknownBookieIdException(e);
}
byte[] data = toString().getBytes(UTF_8);
rm.writeCookie(address.toString(), new Versioned<>(data, version));
rm.writeCookie(address, new Versioned<>(data, version));
}

/**
Expand All @@ -255,9 +255,9 @@ public void writeToRegistrationManager(RegistrationManager rm, ServerConfigurati
public void deleteFromRegistrationManager(RegistrationManager rm,
ServerConfiguration conf,
Version version) throws BookieException {
BookieSocketAddress address = null;
BookieId address = null;
try {
address = Bookie.getBookieAddress(conf);
address = Bookie.getBookieId(conf);
} catch (UnknownHostException e) {
throw new UnknownBookieIdException(e);
}
Expand All @@ -273,13 +273,13 @@ public void deleteFromRegistrationManager(RegistrationManager rm,
* @throws BookieException when fail to delete cookie.
*/
public void deleteFromRegistrationManager(RegistrationManager rm,
BookieSocketAddress address,
BookieId address,
Version version) throws BookieException {
if (!(version instanceof LongVersion)) {
throw new IllegalArgumentException("Invalid version type, expected ZkVersion type");
}

rm.removeCookie(address.toString(), version);
rm.removeCookie(address, version);
}

/**
Expand All @@ -293,7 +293,7 @@ static Builder generateCookie(ServerConfiguration conf)
throws UnknownHostException {
Builder builder = Cookie.newBuilder();
builder.setLayoutVersion(CURRENT_COOKIE_LAYOUT_VERSION);
builder.setBookieHost(Bookie.getBookieAddress(conf).toString());
builder.setBookieHost(Bookie.getBookieId(conf).toString());
builder.setJournalDirs(Joiner.on(',').join(conf.getJournalDirNames()));
builder.setLedgerDirs(encodeDirPaths(conf.getLedgerDirNames()));
return builder;
Expand All @@ -310,7 +310,7 @@ static Builder generateCookie(ServerConfiguration conf)
public static Versioned<Cookie> readFromRegistrationManager(RegistrationManager rm, ServerConfiguration conf)
throws BookieException {
try {
return readFromRegistrationManager(rm, Bookie.getBookieAddress(conf));
return readFromRegistrationManager(rm, Bookie.getBookieId(conf));
} catch (UnknownHostException e) {
throw new UnknownBookieIdException(e);
}
Expand All @@ -325,8 +325,8 @@ public static Versioned<Cookie> readFromRegistrationManager(RegistrationManager
* @throws BookieException when fail to read cookie
*/
public static Versioned<Cookie> readFromRegistrationManager(RegistrationManager rm,
BookieSocketAddress address) throws BookieException {
Versioned<byte[]> cookieData = rm.readCookie(address.toString());
BookieId address) throws BookieException {
Versioned<byte[]> cookieData = rm.readCookie(address);
try {
try (BufferedReader reader = new BufferedReader(
new StringReader(new String(cookieData.getValue(), UTF_8)))) {
Expand Down
Loading