Renkai commented on a change in pull request #9096:
URL: https://github.com/apache/pulsar/pull/9096#discussion_r557939617
##########
File path:
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/LedgerOffloader.java
##########
@@ -22,19 +22,115 @@
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-
+import lombok.ToString;
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.common.annotation.InterfaceAudience;
import org.apache.bookkeeper.common.annotation.InterfaceStability;
+import
org.apache.bookkeeper.mledger.ManagedLedgerException.OffloadNotConsecutiveException;
+import
org.apache.bookkeeper.mledger.ManagedLedgerException.OffloadSegmentClosedException;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats;
import org.apache.pulsar.common.policies.data.OffloadPolicies;
/**
- * Interface for offloading ledgers to long-term storage
+ * Interface for offloading ledgers to long-term storage.
*/
@InterfaceAudience.LimitedPrivate
@InterfaceStability.Evolving
public interface LedgerOffloader {
+ interface SegmentInfo {
+ boolean isClosed();
+
+ OffloadResult result();
+ }
+
+ @ToString
+ class SegmentInfoImpl implements SegmentInfo {
+ public SegmentInfoImpl(UUID uuid, long beginLedger, long beginEntry,
String driverName,
+ Map<String, String> driverMetadata) {
+ this.uuid = uuid;
+ this.beginLedger = beginLedger;
+ this.beginEntry = beginEntry;
+ this.driverName = driverName;
+ this.driverMetadata = driverMetadata;
+ }
+
+
+ public final UUID uuid;
+ public final long beginLedger;
+ public final long beginEntry;
+ public final String driverName;
+ volatile private long endLedger;
+ volatile private long endEntry;
+ volatile boolean closed = false;
+ public final Map<String, String> driverMetadata;
+
+ public boolean isClosed() {
+ return closed;
+ }
+
+ public void closeSegment(long endLedger, long endEntry) {
+ this.endLedger = endLedger;
+ this.endEntry = endEntry;
+ this.closed = true;
+ }
+
+ public OffloadResult result() {
+ return new OffloadResult(beginLedger, beginEntry, endLedger,
endEntry);
+ }
+ }
+
+
+ class OffloadResult {
+ public final long beginLedger;
+ public final long beginEntry;
+ public final long endLedger;
+ public final long endEntry;
+
+ public OffloadResult(long beginLedger, long beginEntry, long
endLedger, long endEntry) {
+ this.beginLedger = beginLedger;
+ this.beginEntry = beginEntry;
+ this.endLedger = endLedger;
+ this.endEntry = endEntry;
+ }
+ }
+
+ /**
+ * Used to store driver info, buffer entries, mark progress, etc.
+ * Create one per second.
+ */
+ interface OffloadHandle {
+
+ /**
+ * return true when both buffer have enough size and ledger/entry id
is next to the current one.
+ * @param size
+ * @return
+ */
+ boolean canOffer(long size);
+
+ default CompletableFuture<Boolean> asyncCanOffer(long size) {
Review comment:
Our current sync method implementation is quite efficient (just compare
two existing number and return), though a bit strange, I still don't think we
should sacrifice the performance to make the implementation conform common
practice.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
