|
48 | 48 | import java.nio.charset.Charset; |
49 | 49 | import java.nio.charset.StandardCharsets; |
50 | 50 | import java.security.GeneralSecurityException; |
| 51 | +import java.time.Duration; |
51 | 52 | import java.util.ArrayList; |
52 | 53 | import java.util.Arrays; |
53 | 54 | import java.util.Collections; |
|
75 | 76 | import java.util.concurrent.atomic.AtomicReference; |
76 | 77 | import java.util.function.Predicate; |
77 | 78 | import lombok.Cleanup; |
| 79 | +import lombok.Data; |
78 | 80 | import lombok.extern.slf4j.Slf4j; |
| 81 | +import org.apache.bookkeeper.client.AsyncCallback; |
79 | 82 | import org.apache.bookkeeper.client.AsyncCallback.AddCallback; |
80 | 83 | import org.apache.bookkeeper.client.BKException; |
81 | 84 | import org.apache.bookkeeper.client.BookKeeper; |
@@ -144,6 +147,117 @@ public Object[][] checkOwnershipFlagProvider() { |
144 | 147 | return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } }; |
145 | 148 | } |
146 | 149 |
|
| 150 | + private void makeAddEntryTimeout(ManagedLedgerImpl ml, AtomicBoolean addEntryFinished) throws Exception { |
| 151 | + LedgerHandle currentLedger = ml.currentLedger; |
| 152 | + final LedgerHandle spyLedgerHandle = spy(currentLedger); |
| 153 | + doAnswer(invocation -> { |
| 154 | + ByteBuf bs = (ByteBuf) invocation.getArguments()[0]; |
| 155 | + AddCallback addCallback = (AddCallback) invocation.getArguments()[1]; |
| 156 | + Object originalContext = invocation.getArguments()[2]; |
| 157 | + currentLedger.asyncAddEntry(bs, (rc, lh, entryId, ctx) -> { |
| 158 | + addEntryFinished.set(true); |
| 159 | + addCallback.addComplete(BKException.Code.TimeoutException, spyLedgerHandle, -1, ctx); |
| 160 | + }, originalContext); |
| 161 | + return null; |
| 162 | + }).when(spyLedgerHandle).asyncAddEntry(any(ByteBuf.class), any(AddCallback.class), any()); |
| 163 | + ml.currentLedger = spyLedgerHandle; |
| 164 | + } |
| 165 | + |
| 166 | + @Data |
| 167 | + private static class DeleteLedgerInfo{ |
| 168 | + volatile boolean hasCalled; |
| 169 | + volatile CompletableFuture<Void> future = new CompletableFuture<>(); |
| 170 | + } |
| 171 | + |
| 172 | + private DeleteLedgerInfo makeDelayIfDoLedgerDelete(LedgerHandle ledger, final AtomicBoolean signal, |
| 173 | + BookKeeper spyBookKeeper) { |
| 174 | + DeleteLedgerInfo deleteLedgerInfo = new DeleteLedgerInfo(); |
| 175 | + doAnswer(invocation -> { |
| 176 | + long ledgerId = (long) invocation.getArguments()[0]; |
| 177 | + AsyncCallback.DeleteCallback originalCb = (AsyncCallback.DeleteCallback) invocation.getArguments()[1]; |
| 178 | + AsyncCallback.DeleteCallback cb = (rc, ctx) -> { |
| 179 | + if (deleteLedgerInfo.hasCalled) { |
| 180 | + deleteLedgerInfo.future.complete(null); |
| 181 | + } |
| 182 | + originalCb.deleteComplete(rc, ctx); |
| 183 | + }; |
| 184 | + Object ctx = invocation.getArguments()[2]; |
| 185 | + if (ledgerId != ledger.getId()){ |
| 186 | + bkc.asyncDeleteLedger(ledgerId, originalCb, ctx); |
| 187 | + } else { |
| 188 | + deleteLedgerInfo.hasCalled = true; |
| 189 | + new Thread(() -> { |
| 190 | + Awaitility.await().atMost(Duration.ofSeconds(60)).until(signal::get); |
| 191 | + bkc.asyncDeleteLedger(ledgerId, cb, ctx); |
| 192 | + }).start(); |
| 193 | + } |
| 194 | + return null; |
| 195 | + }).when(spyBookKeeper).asyncDeleteLedger(any(long.class), any(AsyncCallback.DeleteCallback.class), any()); |
| 196 | + return deleteLedgerInfo; |
| 197 | + } |
| 198 | + |
| 199 | + /*** |
| 200 | + * This test simulates the following problems that can occur when ZK connections are unstable: |
| 201 | + * - add entry timeout |
| 202 | + * - write ZK fail when update ledger info of ML |
| 203 | + * and verifies that ledger info of ML is still correct when the above problems occur. |
| 204 | + */ |
| 205 | + @Test |
| 206 | + public void testLedgerInfoMetaCorrectIfAddEntryTimeOut() throws Exception { |
| 207 | + String mlName = "testLedgerInfoMetaCorrectIfAddEntryTimeOut"; |
| 208 | + BookKeeper spyBookKeeper = spy(bkc); |
| 209 | + ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(metadataStore, spyBookKeeper); |
| 210 | + ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(mlName); |
| 211 | + |
| 212 | + // Make add entry timeout(The data write was actually successful). |
| 213 | + AtomicBoolean addEntryFinished = new AtomicBoolean(false); |
| 214 | + makeAddEntryTimeout(ml, addEntryFinished); |
| 215 | + |
| 216 | + // Make the update operation of ledger info failure when switch ledger. |
| 217 | + metadataStore.failConditional(new MetadataStoreException.BadVersionException(""), (opType, path) -> { |
| 218 | + if (opType == FaultInjectionMetadataStore.OperationType.PUT && addEntryFinished.get() |
| 219 | + && "/managed-ledgers/testLedgerInfoMetaCorrectIfAddEntryTimeOut".equals(path)) { |
| 220 | + return true; |
| 221 | + } |
| 222 | + return false; |
| 223 | + }); |
| 224 | + |
| 225 | + // Make delete ledger is delayed if delete is called. |
| 226 | + AtomicBoolean deleteLedgerDelaySignal = new AtomicBoolean(false); |
| 227 | + DeleteLedgerInfo deleteLedgerInfo = |
| 228 | + makeDelayIfDoLedgerDelete(ml.currentLedger, deleteLedgerDelaySignal, spyBookKeeper); |
| 229 | + |
| 230 | + // Add one entry. |
| 231 | + // - it will fail and trigger ledger switch(we mocked the error). |
| 232 | + // - ledger switch will also fail(we mocked the error). |
| 233 | + try { |
| 234 | + ml.addEntry("1".getBytes(Charset.defaultCharset())); |
| 235 | + fail("Expected the operation of add entry will fail by timeout or ledger fenced."); |
| 236 | + } catch (Exception e){ |
| 237 | + // expected ex. |
| 238 | + } |
| 239 | + |
| 240 | + // Reopen ML. |
| 241 | + try { |
| 242 | + ml.close(); |
| 243 | + fail("Expected the operation of ml close will fail by fenced state."); |
| 244 | + } catch (Exception e){ |
| 245 | + // expected ex. |
| 246 | + } |
| 247 | + ManagedLedgerImpl mlReopened = (ManagedLedgerImpl) factory.open(mlName); |
| 248 | + deleteLedgerDelaySignal.set(true); |
| 249 | + if (deleteLedgerInfo.hasCalled){ |
| 250 | + deleteLedgerInfo.future.join(); |
| 251 | + } |
| 252 | + mlReopened.close(); |
| 253 | + |
| 254 | + // verify: all ledgers in ledger info is worked. |
| 255 | + for (long ledgerId : mlReopened.getLedgersInfo().keySet()){ |
| 256 | + LedgerHandle lh = bkc.openLedger(ledgerId, ml.digestType, ml.getConfig().getPassword()); |
| 257 | + lh.close(); |
| 258 | + } |
| 259 | + } |
| 260 | + |
147 | 261 | @Test |
148 | 262 | public void managedLedgerApi() throws Exception { |
149 | 263 | ManagedLedger ledger = factory.open("my_test_ledger"); |
|
0 commit comments