|
| | 1 | +/** |
| | 2 | + * Licensed to the Apache Software Foundation (ASF) under one |
| | 3 | + * or more contributor license agreements. See the NOTICE file |
| | 4 | + * distributed with this work for additional information |
| | 5 | + * regarding copyright ownership. The ASF licenses this file |
| | 6 | + * to you under the Apache License, Version 2.0 (the |
| | 7 | + * "License"); you may not use this file except in compliance |
| | 8 | + * with the License. You may obtain a copy of the License at |
| | 9 | + * |
| | 10 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| | 11 | + * |
| | 12 | + * Unless required by applicable law or agreed to in writing, |
| | 13 | + * software distributed under the License is distributed on an |
| | 14 | + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| | 15 | + * KIND, either express or implied. See the License for the |
| | 16 | + * specific language governing permissions and limitations |
| | 17 | + * under the License. |
| | 18 | + */ |
| | 19 | +package org.apache.pulsar.broker.s3offload.impl; |
| | 20 | + |
| | 21 | +import com.amazonaws.AmazonClientException; |
| | 22 | +import com.amazonaws.services.s3.AmazonS3; |
| | 23 | +import com.amazonaws.services.s3.model.GetObjectRequest; |
| | 24 | +import com.amazonaws.services.s3.model.ObjectMetadata; |
| | 25 | +import com.amazonaws.services.s3.model.S3Object; |
| | 26 | + |
| | 27 | +import io.netty.buffer.ByteBuf; |
| | 28 | +import io.netty.buffer.PooledByteBufAllocator; |
| | 29 | + |
| | 30 | +import java.io.DataInputStream; |
| | 31 | +import java.io.IOException; |
| | 32 | +import java.util.ArrayList; |
| | 33 | +import java.util.List; |
| | 34 | +import java.util.concurrent.CompletableFuture; |
| | 35 | +import java.util.concurrent.ExecutorService; |
| | 36 | +import java.util.concurrent.ScheduledExecutorService; |
| | 37 | + |
| | 38 | +import org.apache.bookkeeper.client.BKException; |
| | 39 | +import org.apache.bookkeeper.client.api.LastConfirmedAndEntry; |
| | 40 | +import org.apache.bookkeeper.client.api.LedgerEntries; |
| | 41 | +import org.apache.bookkeeper.client.api.LedgerEntry; |
| | 42 | +import org.apache.bookkeeper.client.api.LedgerMetadata; |
| | 43 | +import org.apache.bookkeeper.client.api.ReadHandle; |
| | 44 | +import org.apache.bookkeeper.client.impl.LedgerEntriesImpl; |
| | 45 | +import org.apache.bookkeeper.client.impl.LedgerEntryImpl; |
| | 46 | + |
| | 47 | +import org.apache.pulsar.broker.s3offload.OffloadIndexBlock; |
| | 48 | +import org.apache.pulsar.broker.s3offload.OffloadIndexBlockBuilder; |
| | 49 | +import org.apache.pulsar.broker.s3offload.OffloadIndexEntry; |
| | 50 | +import org.apache.pulsar.broker.s3offload.S3BackedInputStream; |
| | 51 | + |
| | 52 | +import org.slf4j.Logger; |
| | 53 | +import org.slf4j.LoggerFactory; |
| | 54 | + |
| | 55 | +public class S3BackedReadHandleImpl implements ReadHandle { |
| | 56 | + private static final Logger log = LoggerFactory.getLogger(S3BackedReadHandleImpl.class); |
| | 57 | + |
| | 58 | + private final long ledgerId; |
| | 59 | + private final OffloadIndexBlock index; |
| | 60 | + private final S3BackedInputStream inputStream; |
| | 61 | + private final DataInputStream dataStream; |
| | 62 | + private final ExecutorService executor; |
| | 63 | + |
| | 64 | + private S3BackedReadHandleImpl(long ledgerId, OffloadIndexBlock index, |
| | 65 | + S3BackedInputStream inputStream, |
| | 66 | + ExecutorService executor) { |
| | 67 | + this.ledgerId = ledgerId; |
| | 68 | + this.index = index; |
| | 69 | + this.inputStream = inputStream; |
| | 70 | + this.dataStream = new DataInputStream(inputStream); |
| | 71 | + this.executor = executor; |
| | 72 | + } |
| | 73 | + |
| | 74 | + @Override |
| | 75 | + public long getId() { |
| | 76 | + return ledgerId; |
| | 77 | + } |
| | 78 | + |
| | 79 | + @Override |
| | 80 | + public LedgerMetadata getLedgerMetadata() { |
| | 81 | + return index.getLedgerMetadata(); |
| | 82 | + } |
| | 83 | + |
| | 84 | + @Override |
| | 85 | + public CompletableFuture<Void> closeAsync() { |
| | 86 | + CompletableFuture<Void> promise = new CompletableFuture<>(); |
| | 87 | + executor.submit(() -> { |
| | 88 | + try { |
| | 89 | + index.close(); |
| | 90 | + inputStream.close(); |
| | 91 | + promise.complete(null); |
| | 92 | + } catch (IOException t) { |
| | 93 | + promise.completeExceptionally(t); |
| | 94 | + } |
| | 95 | + }); |
| | 96 | + return promise; |
| | 97 | + } |
| | 98 | + |
| | 99 | + @Override |
| | 100 | + public CompletableFuture<LedgerEntries> readAsync(long firstEntry, long lastEntry) { |
| | 101 | + log.debug("Ledger {}: reading {} - {}", getId(), firstEntry, lastEntry); |
| | 102 | + CompletableFuture<LedgerEntries> promise = new CompletableFuture<>(); |
| | 103 | + executor.submit(() -> { |
| | 104 | + if (firstEntry > lastEntry |
| | 105 | + || firstEntry < 0 |
| | 106 | + || lastEntry > getLastAddConfirmed()) { |
| | 107 | + promise.completeExceptionally(new BKException.BKIncorrectParameterException()); |
| | 108 | + return; |
| | 109 | + } |
| | 110 | + long entriesToRead = (lastEntry - firstEntry) + 1; |
| | 111 | + List<LedgerEntry> entries = new ArrayList<LedgerEntry>(); |
| | 112 | + long nextExpectedId = firstEntry; |
| | 113 | + try { |
| | 114 | + OffloadIndexEntry entry = index.getIndexEntryForEntry(firstEntry); |
| | 115 | + inputStream.seek(entry.getDataOffset()); |
| | 116 | + |
| | 117 | + while (entriesToRead > 0) { |
| | 118 | + int length = dataStream.readInt(); |
| | 119 | + if (length < 0) { // hit padding or new block |
| | 120 | + inputStream.seekForward(index.getIndexEntryForEntry(nextExpectedId).getDataOffset()); |
| | 121 | + length = dataStream.readInt(); |
| | 122 | + } |
| | 123 | + long entryId = dataStream.readLong(); |
| | 124 | + |
| | 125 | + if (entryId == nextExpectedId) { |
| | 126 | + ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(length, length); |
| | 127 | + entries.add(LedgerEntryImpl.create(ledgerId, entryId, length, buf)); |
| | 128 | + int toWrite = length; |
| | 129 | + while (toWrite > 0) { |
| | 130 | + toWrite -= buf.writeBytes(dataStream, toWrite); |
| | 131 | + } |
| | 132 | + entriesToRead--; |
| | 133 | + nextExpectedId++; |
| | 134 | + } else if (entryId > lastEntry) { |
| | 135 | + log.info("Expected to read {}, but read {}, which is greater than last entry {}", |
| | 136 | + nextExpectedId, entryId, lastEntry); |
| | 137 | + throw new BKException.BKUnexpectedConditionException(); |
| | 138 | + } else { |
| | 139 | + inputStream.skip(length); |
| | 140 | + } |
| | 141 | + } |
| | 142 | + |
| | 143 | + promise.complete(LedgerEntriesImpl.create(entries)); |
| | 144 | + } catch (Throwable t) { |
| | 145 | + promise.completeExceptionally(t); |
| | 146 | + entries.forEach(LedgerEntry::close); |
| | 147 | + } |
| | 148 | + }); |
| | 149 | + return promise; |
| | 150 | + } |
| | 151 | + |
| | 152 | + @Override |
| | 153 | + public CompletableFuture<LedgerEntries> readUnconfirmedAsync(long firstEntry, long lastEntry) { |
| | 154 | + return readAsync(firstEntry, lastEntry); |
| | 155 | + } |
| | 156 | + |
| | 157 | + @Override |
| | 158 | + public CompletableFuture<Long> readLastAddConfirmedAsync() { |
| | 159 | + return CompletableFuture.completedFuture(getLastAddConfirmed()); |
| | 160 | + } |
| | 161 | + |
| | 162 | + @Override |
| | 163 | + public CompletableFuture<Long> tryReadLastAddConfirmedAsync() { |
| | 164 | + return CompletableFuture.completedFuture(getLastAddConfirmed()); |
| | 165 | + } |
| | 166 | + |
| | 167 | + @Override |
| | 168 | + public long getLastAddConfirmed() { |
| | 169 | + return getLedgerMetadata().getLastEntryId(); |
| | 170 | + } |
| | 171 | + |
| | 172 | + @Override |
| | 173 | + public long getLength() { |
| | 174 | + return getLedgerMetadata().getLength(); |
| | 175 | + } |
| | 176 | + |
| | 177 | + @Override |
| | 178 | + public boolean isClosed() { |
| | 179 | + return getLedgerMetadata().isClosed(); |
| | 180 | + } |
| | 181 | + |
| | 182 | + @Override |
| | 183 | + public CompletableFuture<LastConfirmedAndEntry> readLastAddConfirmedAndEntryAsync(long entryId, |
| | 184 | + long timeOutInMillis, |
| | 185 | + boolean parallel) { |
| | 186 | + CompletableFuture<LastConfirmedAndEntry> promise = new CompletableFuture<>(); |
| | 187 | + promise.completeExceptionally(new UnsupportedOperationException()); |
| | 188 | + return promise; |
| | 189 | + } |
| | 190 | + |
| | 191 | + public static ReadHandle open(ScheduledExecutorService executor, |
| | 192 | + AmazonS3 s3client, String bucket, String key, String indexKey, |
| | 193 | + long ledgerId, int readBufferSize) |
| | 194 | + throws AmazonClientException, IOException { |
| | 195 | + GetObjectRequest req = new GetObjectRequest(bucket, indexKey); |
| | 196 | + try (S3Object obj = s3client.getObject(req)) { |
| | 197 | + OffloadIndexBlockBuilder indexBuilder = OffloadIndexBlockBuilder.create(); |
| | 198 | + OffloadIndexBlock index = indexBuilder.fromStream(obj.getObjectContent()); |
| | 199 | + |
| | 200 | + ObjectMetadata dataMetadata = s3client.getObjectMetadata(bucket, key); // FIXME: this should be part of index |
| | 201 | + S3BackedInputStream inputStream = new S3BackedInputStreamImpl(s3client, bucket, key, |
| | 202 | + dataMetadata.getContentLength(), |
| | 203 | + readBufferSize); |
| | 204 | + return new S3BackedReadHandleImpl(ledgerId, index, inputStream, executor); |
| | 205 | + } |
| | 206 | + } |
| | 207 | +} |
0 commit comments