You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2018/05/22 14:05:18 UTC
[GitHub] eolivelli closed pull request #853: BP-14 WriteFlag DEFERRED_SYNC
Client Side Implementation
eolivelli closed pull request #853: BP-14 WriteFlag DEFERRED_SYNC Client Side Implementation
URL: https://github.com/apache/bookkeeper/pull/853
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java
index 3efff7a21..e1c597951 100644
--- a/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java
+++ b/bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java
@@ -31,6 +31,7 @@
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerHandle;
+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.BookieSocketAddress;
@@ -187,7 +188,8 @@ public static void main(String[] args)
toSend.writeLong(entry);
toSend.writerIndex(toSend.capacity());
bc.addEntry(new BookieSocketAddress(addr, port), ledger, new byte[20],
- entry, ByteBufList.get(toSend), tc, null, BookieProtocol.FLAG_NONE);
+ entry, ByteBufList.get(toSend), tc, null, BookieProtocol.FLAG_NONE,
+ false, WriteFlag.NONE);
}
LOG.info("Waiting for warmup");
tc.waitFor(warmUpCount);
@@ -204,7 +206,8 @@ public static void main(String[] args)
toSend.writerIndex(toSend.capacity());
lc.resetComplete();
bc.addEntry(new BookieSocketAddress(addr, port), ledger, new byte[20],
- entry, ByteBufList.get(toSend), lc, null, BookieProtocol.FLAG_NONE);
+ entry, ByteBufList.get(toSend), lc, null,
+ BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
lc.waitForComplete();
}
long endTime = System.nanoTime();
@@ -222,7 +225,8 @@ public static void main(String[] args)
toSend.writeLong(entry);
toSend.writerIndex(toSend.capacity());
bc.addEntry(new BookieSocketAddress(addr, port), ledger, new byte[20],
- entry, ByteBufList.get(toSend), tc, null, BookieProtocol.FLAG_NONE);
+ entry, ByteBufList.get(toSend), tc, null, BookieProtocol.FLAG_NONE,
+ false, WriteFlag.NONE);
}
tc.waitFor(throughputCount);
endTime = System.currentTimeMillis();
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
index bd0a12d5b..98d365192 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
@@ -33,7 +33,6 @@
import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.IOException;
import java.net.URI;
-import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -871,7 +870,7 @@ public void asyncCreateLedger(final int ensSize, final int writeQuorumSize, fina
}
new LedgerCreateOp(BookKeeper.this, ensSize, writeQuorumSize,
ackQuorumSize, digestType, passwd, cb, ctx,
- customMetadata, EnumSet.noneOf(WriteFlag.class), getStatsLogger())
+ customMetadata, WriteFlag.NONE, getStatsLogger())
.initiate();
} finally {
closeLock.readLock().unlock();
@@ -1075,7 +1074,7 @@ public void asyncCreateLedgerAdv(final int ensSize, final int writeQuorumSize, f
}
new LedgerCreateOp(BookKeeper.this, ensSize, writeQuorumSize,
ackQuorumSize, digestType, passwd, cb, ctx,
- customMetadata, EnumSet.noneOf(WriteFlag.class), getStatsLogger())
+ customMetadata, WriteFlag.NONE, getStatsLogger())
.initiateAdv(-1L);
} finally {
closeLock.readLock().unlock();
@@ -1186,7 +1185,7 @@ public void asyncCreateLedgerAdv(final long ledgerId,
}
new LedgerCreateOp(BookKeeper.this, ensSize, writeQuorumSize,
ackQuorumSize, digestType, passwd, cb, ctx,
- customMetadata, EnumSet.noneOf(WriteFlag.class), getStatsLogger())
+ customMetadata, WriteFlag.NONE, getStatsLogger())
.initiateAdv(ledgerId);
} finally {
closeLock.readLock().unlock();
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
index d6c77de3b..9362b6ad6 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
@@ -24,7 +24,6 @@
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.security.GeneralSecurityException;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
@@ -247,7 +246,7 @@ private void createComplete(int rc, LedgerHandle lh) {
private int builderAckQuorumSize = 2;
private int builderWriteQuorumSize = 2;
private byte[] builderPassword;
- private EnumSet<WriteFlag> builderWriteFlags = EnumSet.noneOf(WriteFlag.class);
+ private EnumSet<WriteFlag> builderWriteFlags = WriteFlag.NONE;
private org.apache.bookkeeper.client.api.DigestType builderDigestType =
org.apache.bookkeeper.client.api.DigestType.CRC32;
private Map<String, byte[]> builderCustomMetadata = Collections.emptyMap();
@@ -262,15 +261,12 @@ public CreateBuilder withEnsembleSize(int ensembleSize) {
return this;
}
+ @Override
public CreateBuilder withWriteFlags(EnumSet<WriteFlag> writeFlags) {
this.builderWriteFlags = writeFlags;
return this;
}
- public CreateBuilder withWriteFlags(WriteFlag... writeFlags) {
- return withWriteFlags(EnumSet.copyOf(Arrays.asList(writeFlags)));
- }
-
@Override
public CreateBuilder withWriteQuorumSize(int writeQuorumSize) {
this.builderWriteQuorumSize = writeQuorumSize;
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
index 3f599e28a..c0514372d 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
@@ -35,6 +35,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookieProtocol;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
@@ -322,7 +323,8 @@ public void readComplete(int rc, LedgerHandle lh,
for (BookieSocketAddress newBookie : newBookies) {
bkc.getBookieClient().addEntry(newBookie, lh.getId(),
lh.getLedgerKey(), entryId, ByteBufList.clone(toSend),
- multiWriteCallback, dataLength, BookieProtocol.FLAG_RECOVERY_ADD);
+ multiWriteCallback, dataLength, BookieProtocol.FLAG_RECOVERY_ADD,
+ false, WriteFlag.NONE);
}
toSend.release();
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
index f72ab8562..c9e44fb67 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
@@ -103,8 +103,19 @@
final BookKeeper bk;
final long ledgerId;
long lastAddPushed;
+
+ /**
+ * Last entryId which has been confirmed to be written durably to the bookies.
+ * This value is used by readers, the the LAC protocol
+ */
volatile long lastAddConfirmed;
+ /**
+ * Next entryId which is expected to move forward during {@link #sendAddSuccessCallbacks() }. This is important
+ * in order to have an ordered sequence of addEntry ackknowledged to the writer
+ */
+ volatile long pendingAddsSequenceHead;
+
long length;
final DigestManager macManager;
final DistributionSchedule distributionSchedule;
@@ -170,6 +181,8 @@
length = 0;
}
+ this.pendingAddsSequenceHead = lastAddConfirmed;
+
this.ledgerId = ledgerId;
if (bk.getConf().getThrottleValue() > 0) {
@@ -1023,7 +1036,7 @@ public void asyncAddEntry(final byte[] data, final int offset, final int length,
public void asyncAddEntry(ByteBuf data, final AddCallback cb, final Object ctx) {
data.retain();
- PendingAddOp op = PendingAddOp.create(this, data, cb, ctx);
+ PendingAddOp op = PendingAddOp.create(this, data, writeFlags, cb, ctx);
doAsyncAddEntry(op);
}
@@ -1090,7 +1103,8 @@ public void asyncAddEntry(final long entryId, final byte[] data, final int offse
*/
void asyncRecoveryAddEntry(final byte[] data, final int offset, final int length,
final AddCallback cb, final Object ctx) {
- PendingAddOp op = PendingAddOp.create(this, Unpooled.wrappedBuffer(data, offset, length), cb, ctx)
+ PendingAddOp op = PendingAddOp.create(this, Unpooled.wrappedBuffer(data, offset, length),
+ writeFlags, cb, ctx)
.enableRecoveryAdd();
doAsyncAddEntry(op);
}
@@ -1678,17 +1692,20 @@ void sendAddSuccessCallbacks() {
return;
}
// Check if it is the next entry in the sequence.
- if (pendingAddOp.entryId != 0 && pendingAddOp.entryId != lastAddConfirmed + 1) {
+ if (pendingAddOp.entryId != 0 && pendingAddOp.entryId != pendingAddsSequenceHead + 1) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Head of the queue entryId: {} is not lac: {} + 1", pendingAddOp.entryId,
- lastAddConfirmed);
+ LOG.debug("Head of the queue entryId: {} is not the expected value: {}", pendingAddOp.entryId,
+ pendingAddsSequenceHead + 1);
}
return;
}
pendingAddOps.remove();
explicitLacFlushPolicy.updatePiggyBackedLac(lastAddConfirmed);
- lastAddConfirmed = pendingAddOp.entryId;
+ pendingAddsSequenceHead = pendingAddOp.entryId;
+ if (!writeFlags.contains(WriteFlag.DEFERRED_SYNC)) {
+ this.lastAddConfirmed = pendingAddsSequenceHead;
+ }
pendingAddOp.submitCallback(BKException.Code.OK);
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
index f67e1d674..70e94300c 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
@@ -179,7 +179,7 @@ public void asyncAddEntry(final long entryId, final byte[] data, final int offse
private void asyncAddEntry(final long entryId, ByteBuf data,
final AddCallbackWithLatency cb, final Object ctx) {
- PendingAddOp op = PendingAddOp.create(this, data, cb, ctx);
+ PendingAddOp op = PendingAddOp.create(this, data, writeFlags, cb, ctx);
op.setEntryId(entryId);
if ((entryId <= this.lastAddConfirmed) || pendingAddOps.contains(op)) {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java
index 3956c6867..d414b7b71 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java
@@ -112,6 +112,7 @@ public void readLastConfirmedDataComplete(int rc, RecoveryData data) {
synchronized (lh) {
lh.lastAddPushed = lh.lastAddConfirmed = data.getLastAddConfirmed();
lh.length = data.getLength();
+ lh.pendingAddsSequenceHead = lh.lastAddConfirmed;
startEntryToRead = endEntryToRead = lh.lastAddConfirmed;
}
// keep a copy of ledger metadata before proceeding
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
index 61037f0d0..8e30abfbf 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingAddOp.java
@@ -28,12 +28,13 @@
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
import io.netty.util.ReferenceCountUtil;
+import java.util.EnumSet;
import java.util.Map;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
-
import org.apache.bookkeeper.client.AsyncCallback.AddCallbackWithLatency;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
import org.apache.bookkeeper.stats.Counter;
@@ -79,10 +80,10 @@
int pendingWriteRequests;
boolean callbackTriggered;
boolean hasRun;
-
+ EnumSet<WriteFlag> writeFlags;
boolean allowFailFast = false;
-
- static PendingAddOp create(LedgerHandle lh, ByteBuf payload, AddCallbackWithLatency cb, Object ctx) {
+ static PendingAddOp create(LedgerHandle lh, ByteBuf payload, EnumSet<WriteFlag> writeFlags,
+ AddCallbackWithLatency cb, Object ctx) {
PendingAddOp op = RECYCLER.get();
op.lh = lh;
op.isRecoveryAdd = false;
@@ -104,6 +105,7 @@ static PendingAddOp create(LedgerHandle lh, ByteBuf payload, AddCallbackWithLate
op.requestTimeNanos = Long.MAX_VALUE;
op.allowFailFast = false;
op.qwcLatency = 0;
+ op.writeFlags = writeFlags;
return op;
}
@@ -137,7 +139,7 @@ void sendWriteRequest(int bookieIndex) {
int flags = isRecoveryAdd ? FLAG_RECOVERY_ADD | FLAG_HIGH_PRIORITY : FLAG_NONE;
lh.bk.getBookieClient().addEntry(lh.metadata.currentEnsemble.get(bookieIndex), lh.ledgerId, lh.ledgerKey,
- entryId, toSend, this, bookieIndex, flags, allowFailFast);
+ entryId, toSend, this, bookieIndex, flags, allowFailFast, lh.writeFlags);
++pendingWriteRequests;
}
@@ -307,6 +309,10 @@ public void writeComplete(int rc, long ledgerId, long entryId, BookieSocketAddre
// bookie client is closed.
lh.errorOutPendingAdds(rc);
return;
+ case BKException.Code.IllegalOpException:
+ // illegal operation requested, like using unsupported feature in v2 protocol
+ lh.handleUnrecoverableErrorDuringAdd(rc);
+ return;
case BKException.Code.LedgerFencedException:
LOG.warn("Fencing exception on write: L{} E{} on {}",
ledgerId, entryId, addr);
@@ -453,6 +459,7 @@ private void recyclePendAddOpObject() {
callbackTriggered = false;
hasRun = false;
allowFailFast = false;
+ writeFlags = null;
recyclerHandle.recycle(this);
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
index eef0e70ea..f2433edb9 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
@@ -21,7 +21,6 @@
package org.apache.bookkeeper.client;
import java.security.GeneralSecurityException;
-import java.util.EnumSet;
import java.util.Map;
import java.util.concurrent.RejectedExecutionException;
@@ -77,7 +76,7 @@ public String toString() {
ReadOnlyLedgerHandle(BookKeeper bk, long ledgerId, LedgerMetadata metadata,
DigestType digestType, byte[] password, boolean watch)
throws GeneralSecurityException, NumberFormatException {
- super(bk, ledgerId, metadata, digestType, password, EnumSet.noneOf(WriteFlag.class));
+ super(bk, ledgerId, metadata, digestType, password, WriteFlag.NONE);
if (watch) {
bk.getLedgerManager().registerLedgerMetadataListener(ledgerId, this);
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java
index cd5fd3d7c..a0b67e065 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java
@@ -20,6 +20,8 @@
*/
package org.apache.bookkeeper.client.api;
+import java.util.Arrays;
+import java.util.EnumSet;
import java.util.Map;
import org.apache.bookkeeper.common.annotation.InterfaceAudience.Public;
import org.apache.bookkeeper.common.annotation.InterfaceStability.Unstable;
@@ -72,6 +74,26 @@
*/
CreateBuilder withPassword(byte[] password);
+ /**
+ * Set write flags. Write flags specify the behaviour of writes
+ *
+ * @param writeFlags the flags
+ *
+ * @return the builder itself
+ */
+ CreateBuilder withWriteFlags(EnumSet<WriteFlag> writeFlags);
+
+ /**
+ * Set write flags. Write flags specify the behaviour of writes
+ *
+ * @param writeFlags the flags
+ *
+ * @return the builder itself
+ */
+ default CreateBuilder withWriteFlags(WriteFlag ... writeFlags) {
+ return withWriteFlags(EnumSet.copyOf(Arrays.asList(writeFlags)));
+ }
+
/**
* Set a map a custom data to be attached to the ledger. The application is responsible for the semantics of these
* data.
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/WriteFlag.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/WriteFlag.java
index a680b8c62..30199c268 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/WriteFlag.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/WriteFlag.java
@@ -35,6 +35,13 @@
*/
DEFERRED_SYNC(0x1 << 0);
+ /**
+ * No flag is set, use default behaviour.
+ */
+ public static final EnumSet<WriteFlag> NONE = EnumSet.noneOf(WriteFlag.class);
+
+ private static final EnumSet<WriteFlag> ONLY_DEFERRED_SYNC = EnumSet.of(DEFERRED_SYNC);
+
private final int value;
WriteFlag(int value) {
@@ -49,9 +56,9 @@
*/
public static EnumSet<WriteFlag> getWriteFlags(int flagValue) {
if ((flagValue & DEFERRED_SYNC.value) == DEFERRED_SYNC.value) {
- return EnumSet.of(DEFERRED_SYNC);
+ return ONLY_DEFERRED_SYNC;
}
- return EnumSet.noneOf(WriteFlag.class);
+ return WriteFlag.NONE;
}
/**
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
index 31c12fcbd..3dd837cd6 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieClient.java
@@ -35,6 +35,7 @@
import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.IOException;
+import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
@@ -49,6 +50,7 @@
import org.apache.bookkeeper.auth.ClientAuthProvider;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.bookkeeper.common.util.SafeRunnable;
import org.apache.bookkeeper.conf.ClientConfiguration;
@@ -240,17 +242,6 @@ public String toString() {
}
}
- public void addEntry(final BookieSocketAddress addr,
- final long ledgerId,
- final byte[] masterKey,
- final long entryId,
- final ByteBufList toSend,
- final WriteCallback cb,
- final Object ctx,
- final int options) {
- addEntry(addr, ledgerId, masterKey, entryId, toSend, cb, ctx, options, false);
- }
-
public void addEntry(final BookieSocketAddress addr,
final long ledgerId,
final byte[] masterKey,
@@ -259,7 +250,8 @@ public void addEntry(final BookieSocketAddress addr,
final WriteCallback cb,
final Object ctx,
final int options,
- final boolean allowFastFail) {
+ final boolean allowFastFail,
+ final EnumSet<WriteFlag> writeFlags) {
final PerChannelBookieClientPool client = lookupClient(addr);
if (client == null) {
completeAdd(getRc(BKException.Code.BookieHandleNotAvailableException),
@@ -273,7 +265,7 @@ public void addEntry(final BookieSocketAddress addr,
client.obtain(ChannelReadyForAddEntryCallback.create(
this, toSend, ledgerId, entryId, addr,
- ctx, cb, options, masterKey, allowFastFail),
+ ctx, cb, options, masterKey, allowFastFail, writeFlags),
ledgerId);
}
@@ -310,11 +302,13 @@ public void safeRun() {
private int options;
private byte[] masterKey;
private boolean allowFastFail;
+ private EnumSet<WriteFlag> writeFlags;
static ChannelReadyForAddEntryCallback create(
BookieClient bookieClient, ByteBufList toSend, long ledgerId,
long entryId, BookieSocketAddress addr, Object ctx,
- WriteCallback cb, int options, byte[] masterKey, boolean allowFastFail) {
+ WriteCallback cb, int options, byte[] masterKey, boolean allowFastFail,
+ EnumSet<WriteFlag> writeFlags) {
ChannelReadyForAddEntryCallback callback = RECYCLER.get();
callback.bookieClient = bookieClient;
callback.toSend = toSend;
@@ -326,6 +320,7 @@ static ChannelReadyForAddEntryCallback create(
callback.options = options;
callback.masterKey = masterKey;
callback.allowFastFail = allowFastFail;
+ callback.writeFlags = writeFlags;
return callback;
}
@@ -336,7 +331,7 @@ public void operationComplete(final int rc,
bookieClient.completeAdd(rc, ledgerId, entryId, addr, cb, ctx);
} else {
pcbc.addEntry(ledgerId, masterKey, entryId,
- toSend, cb, ctx, options, allowFastFail);
+ toSend, cb, ctx, options, allowFastFail, writeFlags);
}
toSend.release();
@@ -367,7 +362,7 @@ public void recycle() {
options = -1;
masterKey = null;
allowFastFail = false;
-
+ writeFlags = null;
recyclerHandle.recycle(this);
}
}
@@ -565,7 +560,9 @@ public void writeComplete(int rc, long ledger, long entry, BookieSocketAddress a
for (int i = 0; i < 100000; i++) {
counter.inc();
- bc.addEntry(addr, ledger, new byte[0], i, ByteBufList.get(Unpooled.wrappedBuffer(hello)), cb, counter, 0);
+ bc.addEntry(addr, ledger, new byte[0], i,
+ ByteBufList.get(Unpooled.wrappedBuffer(hello)), cb, counter, 0, false,
+ WriteFlag.NONE);
}
counter.wait(0);
System.out.println("Total = " + counter.total());
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
index f6ebbfade..4d69acf85 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
@@ -70,6 +70,7 @@
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
@@ -87,6 +88,7 @@
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeperClientStats;
import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo;
+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.BookieSocketAddress;
@@ -579,19 +581,23 @@ void writeLac(final long ledgerId, final byte[] masterKey, final long lac, ByteB
* Write callback
* @param ctx
* Write callback context
- * @param options
- * Add options
+ * @param allowFastFail
+ * allowFastFail flag
+ * @param writeFlags
+ * WriteFlags
*/
void addEntry(final long ledgerId, byte[] masterKey, final long entryId, ByteBufList toSend, WriteCallback cb,
- Object ctx, final int options) {
- addEntry(ledgerId, masterKey, entryId, toSend, cb, ctx, options, false);
- }
-
- void addEntry(final long ledgerId, byte[] masterKey, final long entryId, ByteBufList toSend, WriteCallback cb,
- Object ctx, final int options, boolean allowFastFail) {
+ Object ctx, final int options, boolean allowFastFail, final EnumSet<WriteFlag> writeFlags) {
Object request = null;
CompletionKey completionKey = null;
if (useV2WireProtocol) {
+ if (writeFlags.contains(WriteFlag.DEFERRED_SYNC)) {
+ LOG.error("invalid writeflags {} for v2 protocol", writeFlags);
+ executor.executeOrdered(ledgerId, () -> {
+ cb.writeComplete(BKException.Code.IllegalOpException, ledgerId, entryId, addr, ctx);
+ });
+ return;
+ }
completionKey = acquireV2Key(ledgerId, entryId, OperationType.ADD_ENTRY);
request = BookieProtocol.AddRequest.create(
BookieProtocol.CURRENT_PROTOCOL_VERSION, ledgerId, entryId,
@@ -627,6 +633,10 @@ void addEntry(final long ledgerId, byte[] masterKey, final long entryId, ByteBuf
addBuilder.setFlag(AddRequest.Flag.RECOVERY_ADD);
}
+ if (!writeFlags.isEmpty()) {
+ // add flags only if needed, in order to be able to talk with old bookies
+ addBuilder.setWriteFlags(WriteFlag.getWriteFlagsValue(writeFlags));
+ }
request = Request.newBuilder()
.setHeader(headerBuilder)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java
index 578d666ae..fe68da42a 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java
@@ -110,7 +110,7 @@ public void writeComplete(int rc, long ledgerId, long entryId,
if (addRequest.hasWriteFlags()) {
writeFlags = WriteFlag.getWriteFlags(addRequest.getWriteFlags());
} else {
- writeFlags = EnumSet.noneOf(WriteFlag.class);
+ writeFlags = WriteFlag.NONE;
}
final boolean ackBeforeSync = writeFlags.contains(WriteFlag.DEFERRED_SYNC);
StatusCode status = null;
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperTest.java
index 1acc6fd51..f1521f4e0 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperTest.java
@@ -20,6 +20,7 @@
*/
package org.apache.bookkeeper.client;
+import static org.apache.bookkeeper.common.concurrent.FutureUtils.result;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -39,7 +40,10 @@
import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
import org.apache.bookkeeper.client.BKException.BKBookieHandleNotAvailableException;
+import org.apache.bookkeeper.client.BKException.BKIllegalOpException;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.api.WriteFlag;
+import org.apache.bookkeeper.client.api.WriteHandle;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.zookeeper.KeeperException.ConnectionLossException;
@@ -779,4 +783,22 @@ public void readComplete(int rc, LedgerHandle lh,
latch.await();
bkc.close();
}
+
+ @Test(expected = BKIllegalOpException.class)
+ public void testCannotUseWriteFlagsOnV2Protocol() throws Exception {
+ ClientConfiguration conf = new ClientConfiguration(baseClientConf);
+ conf.setUseV2WireProtocol(true);
+ try (BookKeeperTestClient bkc = new BookKeeperTestClient(conf);) {
+ try (WriteHandle wh = result(bkc.newCreateLedgerOp()
+ .withEnsembleSize(3)
+ .withWriteQuorumSize(3)
+ .withAckQuorumSize(2)
+ .withPassword("".getBytes())
+ .withWriteFlags(WriteFlag.DEFERRED_SYNC)
+ .execute())) {
+ result(wh.appendAsync("test".getBytes()));
+ }
+ }
+ }
+
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/DeferredSyncTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/DeferredSyncTest.java
new file mode 100644
index 000000000..4bd5a8cb8
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/DeferredSyncTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.apache.bookkeeper.common.concurrent.FutureUtils.result;
+import static org.junit.Assert.assertEquals;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.bookkeeper.client.api.WriteFlag;
+import org.apache.bookkeeper.client.api.WriteHandle;
+import org.junit.Test;
+
+/**
+ * Client side tests on deferred sync write flag.
+ */
+public class DeferredSyncTest extends MockBookKeeperTestCase {
+
+ static final byte[] PASSWORD = "password".getBytes();
+ static final ByteBuf DATA = Unpooled.wrappedBuffer("foobar".getBytes());
+ static final int NUM_ENTRIES = 100;
+
+ @Test
+ public void testAddEntryLastAddConfirmedDoesNotAdvance() throws Exception {
+ try (WriteHandle wh = result(newCreateLedgerOp()
+ .withEnsembleSize(3)
+ .withWriteQuorumSize(3)
+ .withAckQuorumSize(2)
+ .withPassword(PASSWORD)
+ .withWriteFlags(WriteFlag.DEFERRED_SYNC)
+ .execute())) {
+ for (int i = 0; i < NUM_ENTRIES - 1; i++) {
+ result(wh.appendAsync(DATA));
+ }
+ long lastEntryID = result(wh.appendAsync(DATA));
+ assertEquals(NUM_ENTRIES - 1, lastEntryID);
+ LedgerHandle lh = (LedgerHandle) wh;
+ assertEquals(NUM_ENTRIES - 1, lh.getLastAddPushed());
+ assertEquals(-1, lh.getLastAddConfirmed());
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
index 433591c98..b8536142d 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
@@ -36,6 +36,7 @@
import java.security.GeneralSecurityException;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -49,7 +50,7 @@
import org.apache.bookkeeper.client.BKException.BKDigestMatchException;
import org.apache.bookkeeper.client.BKException.Code;
-import org.apache.bookkeeper.client.LedgerCreateOp.CreateBuilderImpl;
+import org.apache.bookkeeper.client.api.CreateBuilder;
import org.apache.bookkeeper.client.api.DeleteBuilder;
import org.apache.bookkeeper.client.api.OpenBuilder;
import org.apache.bookkeeper.common.util.OrderedExecutor;
@@ -213,7 +214,7 @@ protected void setBookkeeperConfig(ClientConfiguration config) {
when(bk.getConf()).thenReturn(config);
}
- protected CreateBuilderImpl newCreateLedgerOp() {
+ protected CreateBuilder newCreateLedgerOp() {
return new LedgerCreateOp.CreateBuilderImpl(bk);
}
@@ -522,13 +523,7 @@ protected void setupBookieClientAddEntry() {
anyLong(), any(byte[].class),
anyLong(), any(ByteBufList.class),
any(BookkeeperInternalCallbacks.WriteCallback.class),
- any(), anyInt());
-
- stub.when(bookieClient).addEntry(any(BookieSocketAddress.class),
- anyLong(), any(byte[].class),
- anyLong(), any(ByteBufList.class),
- any(BookkeeperInternalCallbacks.WriteCallback.class),
- any(), anyInt(), anyBoolean());
+ any(), anyInt(), anyBoolean(), any(EnumSet.class));
}
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockLedgerHandle.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockLedgerHandle.java
index 838ed5be9..a0d339c29 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockLedgerHandle.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockLedgerHandle.java
@@ -27,7 +27,6 @@
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.EnumSet;
import java.util.Enumeration;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
@@ -61,7 +60,7 @@
MockLedgerHandle(MockBookKeeper bk, long id, DigestType digest, byte[] passwd) throws GeneralSecurityException {
super(bk, id, new LedgerMetadata(3, 3, 2, DigestType.MAC, "".getBytes()), DigestType.MAC, "".getBytes(),
- EnumSet.noneOf(WriteFlag.class));
+ WriteFlag.NONE);
this.bk = bk;
this.id = id;
this.digest = digest;
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java
index a6cd90202..b21386b0e 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java
@@ -42,6 +42,7 @@
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.bookie.InterleavedLedgerStorage;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
@@ -434,7 +435,7 @@ public void writeComplete(int rc, long ledgerId, long entryId, BookieSocketAddre
addSuccess.set(BKException.Code.OK == rc);
addLatch.countDown();
}
- }, 0, BookieProtocol.FLAG_NONE);
+ }, 0, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
addLatch.await();
assertTrue("add entry 14 should succeed", addSuccess.get());
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java
index e9af76ee5..f428b1f49 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java
@@ -29,6 +29,7 @@
import io.netty.buffer.Unpooled;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.bookkeeper.client.BKException.Code;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.junit.Before;
import org.junit.Test;
@@ -61,7 +62,7 @@ public void setup() {
public void testExecuteAfterCancelled() {
AtomicInteger rcHolder = new AtomicInteger(-0xdead);
PendingAddOp op = PendingAddOp.create(
- lh, payload, (rc, handle, entryId, qwcLatency, ctx) -> {
+ lh, payload, WriteFlag.NONE, (rc, handle, entryId, qwcLatency, ctx) -> {
rcHolder.set(rc);
}, null);
assertSame(lh, op.lh);
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerFragmentReplication.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerFragmentReplication.java
index c620cc16e..883dc9c91 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerFragmentReplication.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerFragmentReplication.java
@@ -26,7 +26,6 @@
import com.google.common.collect.Sets;
import java.util.ArrayList;
-import java.util.EnumSet;
import java.util.Enumeration;
import java.util.Map.Entry;
import java.util.Set;
@@ -248,7 +247,7 @@ public boolean isClosed() {
}
};
LedgerHandle lh = new LedgerHandle(bkc, 0, metadata, TEST_DIGEST_TYPE,
- TEST_PSSWD, EnumSet.noneOf(WriteFlag.class));
+ TEST_PSSWD, WriteFlag.NONE);
testSplitIntoSubFragments(10, 21, -1, 1, lh);
testSplitIntoSubFragments(10, 21, 20, 1, lh);
testSplitIntoSubFragments(0, 0, 10, 1, lh);
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/BookKeeperBuildersTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/BookKeeperBuildersTest.java
index cac76d41b..b0385b108 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/BookKeeperBuildersTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/BookKeeperBuildersTest.java
@@ -218,7 +218,7 @@ public void testDefaultWriteFlagsEmpty() throws Exception {
assertEquals(writeQuorumSize, metadata.getWriteQuorumSize());
assertArrayEquals(password, metadata.getPassword());
LedgerHandle lh = (LedgerHandle) writer;
- assertEquals(EnumSet.noneOf(WriteFlag.class), lh.getWriteFlags());
+ assertEquals(WriteFlag.NONE, lh.getWriteFlags());
}
@Test
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/WriteFlagTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/WriteFlagTest.java
index c51bdf5bc..746535a06 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/WriteFlagTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/WriteFlagTest.java
@@ -41,7 +41,7 @@ public void testGetWriteFlagsDeferredSync() {
@Test
public void testGetWriteFlagsNone() {
- assertEquals(EnumSet.noneOf(WriteFlag.class),
+ assertEquals(WriteFlag.NONE,
WriteFlag.getWriteFlags(NONE));
}
@@ -52,7 +52,7 @@ public void testGetWriteFlagsValueNull() {
@Test
public void testGetWriteFlagsValueEmpty() {
- assertEquals(0, WriteFlag.getWriteFlagsValue(EnumSet.noneOf(WriteFlag.class)));
+ assertEquals(0, WriteFlag.getWriteFlagsValue(WriteFlag.NONE));
}
@Test
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
index 7acdb868b..7ff638675 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
@@ -40,6 +40,7 @@
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BKException.Code;
import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo;
+import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -158,7 +159,7 @@ public void testWriteGaps() throws Exception {
BookieClient bc = new BookieClient(new ClientConfiguration(), eventLoopGroup, executor,
scheduler, NullStatsLogger.INSTANCE);
ByteBufList bb = createByteBuffer(1, 1, 1);
- bc.addEntry(addr, 1, passwd, 1, bb, wrcb, arc, BookieProtocol.FLAG_NONE);
+ bc.addEntry(addr, 1, passwd, 1, bb, wrcb, arc, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
synchronized (arc) {
arc.wait(1000);
assertEquals(0, arc.rc);
@@ -168,16 +169,16 @@ public void testWriteGaps() throws Exception {
assertEquals(1, arc.entry.getInt());
}
bb = createByteBuffer(2, 1, 2);
- bc.addEntry(addr, 1, passwd, 2, bb, wrcb, null, BookieProtocol.FLAG_NONE);
+ bc.addEntry(addr, 1, passwd, 2, bb, wrcb, null, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
bb = createByteBuffer(3, 1, 3);
- bc.addEntry(addr, 1, passwd, 3, bb, wrcb, null, BookieProtocol.FLAG_NONE);
+ bc.addEntry(addr, 1, passwd, 3, bb, wrcb, null, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
bb = createByteBuffer(5, 1, 5);
- bc.addEntry(addr, 1, passwd, 5, bb, wrcb, null, BookieProtocol.FLAG_NONE);
+ bc.addEntry(addr, 1, passwd, 5, bb, wrcb, null, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
bb = createByteBuffer(7, 1, 7);
- bc.addEntry(addr, 1, passwd, 7, bb, wrcb, null, BookieProtocol.FLAG_NONE);
+ bc.addEntry(addr, 1, passwd, 7, bb, wrcb, null, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
synchronized (notifyObject) {
bb = createByteBuffer(11, 1, 11);
- bc.addEntry(addr, 1, passwd, 11, bb, wrcb, notifyObject, BookieProtocol.FLAG_NONE);
+ bc.addEntry(addr, 1, passwd, 11, bb, wrcb, notifyObject, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
notifyObject.wait();
}
synchronized (arc) {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services