Skip to content

Commit 9dd87e2

Browse files
horizonzyhorizonzy
committed
code clean
1 parent 7e6c007 commit 9dd87e2

File tree

5 files changed

+28
-82
lines changed

5 files changed

+28
-82
lines changed

‎managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@ public class ManagedLedgerConfig {
8383
private boolean cacheEvictionByMarkDeletedPosition = false;
8484
private boolean supportTwoPhaseDeletion = false;
8585
private int archiveDataLimitSize = 500;
86-
private int deleteIntervalSeconds = 60000;
86+
private int deleteIntervalSeconds = 60;
8787

8888
public boolean isCreateIfMissing() {
8989
return createIfMissing;

‎managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

+4-3
Original file line numberDiff line numberDiff line change
@@ -341,8 +341,8 @@ public ManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper
341341
}
342342
this.inactiveLedgerRollOverTimeMs = config.getInactiveLedgerRollOverTimeMs();
343343
this.managedTrash = config.isSupportTwoPhaseDeletion()
344-
? new ManagedTrashImpl(ManagedTrash.ManagedType.MANAGED_LEDGER, name, metadataStore, config, scheduledExecutor,
345-
executor, bookKeeper) : new ManagedTrashDisableImpl();
344+
? new ManagedTrashImpl(ManagedTrash.ManagedType.MANAGED_LEDGER, name, metadataStore, config,
345+
scheduledExecutor, executor, bookKeeper) : new ManagedTrashDisableImpl();
346346
}
347347

348348
synchronized void initialize(final ManagedLedgerInitializeLedgerCallback callback, final Object ctx) {
@@ -2688,7 +2688,8 @@ private CompletableFuture<?> asyncUpdateTrashData(Collection<Long> deletableLedg
26882688
managedTrash.appendLedgerTrashData(ledgerId, null, ManagedTrash.LedgerType.LEDGER);
26892689
}
26902690
for (Long ledgerId : deletableOffloadedLedgerIds) {
2691-
managedTrash.appendLedgerTrashData(ledgerId, ledgers.get(ledgerId), ManagedTrash.LedgerType.OFFLOAD_LEDGER);
2691+
managedTrash.appendLedgerTrashData(ledgerId, ledgers.get(ledgerId),
2692+
ManagedTrash.LedgerType.OFFLOAD_LEDGER);
26922693
}
26932694
future.complete(null);
26942695
} catch (ManagedLedgerException e) {

‎managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedTrashImpl.java

+7-7
Original file line numberDiff line numberDiff line change
@@ -151,7 +151,7 @@ public CompletableFuture<Void> initialize() {
151151
return;
152152
}
153153
if (res.isEmpty()) {
154-
STATE_UPDATER.set(this, State.INITIALIZED);
154+
STATE_UPDATER.set(this, State.Initialized);
155155
checkTrashPersistTask =
156156
scheduledExecutor.scheduleAtFixedRate(safeRun(this::persistTrashIfNecessary), 30L, 30L,
157157
TimeUnit.MINUTES);
@@ -167,7 +167,7 @@ public CompletableFuture<Void> initialize() {
167167
checkTrashPersistTask =
168168
scheduledExecutor.scheduleAtFixedRate(safeRun(this::persistTrashIfNecessary), 30L, 30L,
169169
TimeUnit.MINUTES);
170-
STATE_UPDATER.set(this, State.INITIALIZED);
170+
STATE_UPDATER.set(this, State.Initialized);
171171
triggerDeleteInBackground();
172172
} catch (InvalidProtocolBufferException exc) {
173173
future.completeExceptionally(ManagedLedgerException.getManagedLedgerException(exc));
@@ -197,7 +197,7 @@ private int calculateArchiveCount() {
197197
public void appendLedgerTrashData(long ledgerId, LedgerInfo context, LedgerType type)
198198
throws ManagedLedgerException {
199199
State state = STATE_UPDATER.get(this);
200-
if (state != State.INITIALIZED) {
200+
if (state != State.Initialized) {
201201
throw ManagedLedgerException.getManagedLedgerException(new IllegalStateException(
202202
String.format("[%s] is not initialized, current state: %s", name(), state)));
203203
}
@@ -230,7 +230,7 @@ public void asyncUpdateTrashDataInBackground(CompletableFuture<?> future) {
230230

231231
private void doAsyncUpdateTrashData(CompletableFuture<?> future) {
232232
State state = STATE_UPDATER.get(this);
233-
if (state != State.INITIALIZED) {
233+
if (state != State.Initialized) {
234234
future.completeExceptionally(ManagedLedgerException.getManagedLedgerException(new IllegalStateException(
235235
String.format("[%s] is not initialized, current state: %s", name(), state))));
236236
return;
@@ -306,7 +306,7 @@ public void triggerDeleteInBackground() {
306306

307307
private void triggerDelete() {
308308
State state = STATE_UPDATER.get(this);
309-
if (state != State.INITIALIZED) {
309+
if (state != State.Initialized) {
310310
log.warn("[{}] is not initialized, current state: {}", name(), state);
311311
return;
312312
}
@@ -460,7 +460,7 @@ private void asyncUpdateArchiveData(CompletableFuture<?> future) {
460460
log.info("[{}] Start async update archive data", name());
461461

462462
State state = STATE_UPDATER.get(this);
463-
if (state != State.INITIALIZED) {
463+
if (state != State.Initialized) {
464464
future.completeExceptionally(ManagedLedgerException.getManagedLedgerException(new IllegalStateException(
465465
String.format("[%s] is not initialized, current state: %s", name(), state))));
466466
return;
@@ -754,7 +754,7 @@ public int compareTo(TrashKey other) {
754754

755755
public enum State {
756756
None,
757-
INITIALIZED,
757+
Initialized,
758758
Closed,
759759
}
760760

‎managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedTrashTest.java

+13-66
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
import java.lang.reflect.Field;
2424
import java.nio.charset.Charset;
2525
import java.util.ArrayList;
26-
import java.util.EnumSet;
2726
import java.util.HashMap;
2827
import java.util.List;
2928
import java.util.Map;
@@ -43,16 +42,10 @@
4342
import org.apache.bookkeeper.mledger.ManagedTrash;
4443
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
4544
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
46-
import org.apache.pulsar.common.util.FutureUtil;
4745
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
48-
import org.apache.pulsar.metadata.api.MetadataStoreException;
49-
import org.apache.pulsar.metadata.api.Notification;
50-
import org.apache.pulsar.metadata.api.NotificationType;
5146
import org.apache.pulsar.metadata.api.Stat;
52-
import org.apache.pulsar.metadata.api.extended.CreateOption;
5347
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
5448
import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore;
55-
import org.apache.pulsar.metadata.impl.LocalMemoryMetadataStore;
5649
import org.awaitility.Awaitility;
5750
import org.testng.Assert;
5851
import org.testng.annotations.Test;
@@ -72,66 +65,20 @@ public class ManagedTrashTest extends MockedBookKeeperTestCase {
7265

7366
@Override
7467
protected void setUpTestCase() throws Exception {
75-
MetadataStoreExtended metadataStoreExtended =
76-
new LocalMemoryMetadataStore("memory:local", MetadataStoreConfig.builder().build()) {
77-
@Override
78-
public CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Long> optExpectedVersion,
79-
EnumSet<CreateOption> options) {
80-
if (!isValidPath(path)) {
81-
return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
82-
}
83-
synchronized (map) {
84-
boolean hasVersion = optExpectedVersion.isPresent();
85-
int expectedVersion = optExpectedVersion.orElse(-1L).intValue();
86-
87-
if (options.contains(CreateOption.Sequential)) {
88-
path += Long.toString(sequentialIdGenerator.getAndIncrement());
89-
}
90-
91-
long now = System.currentTimeMillis();
92-
93-
if (hasVersion && expectedVersion == -1) {
94-
Value newValue = new Value(0, data, now, now, options.contains(CreateOption.Ephemeral));
95-
Value existingValue = map.putIfAbsent(path, newValue);
96-
if (existingValue != null) {
97-
return FutureUtils.exception(new MetadataStoreException.BadVersionException(""));
98-
} else {
99-
persistedData.put(path, data);
100-
receivedNotification(new Notification(NotificationType.Created, path));
101-
notifyParentChildrenChanged(path);
102-
return FutureUtils.value(new Stat(path, 0, now, now, newValue.isEphemeral(), true));
103-
}
104-
} else {
105-
Value existingValue = map.get(path);
106-
long existingVersion = existingValue != null ? existingValue.getVersion() : -1;
107-
if (hasVersion && expectedVersion != existingVersion) {
108-
return FutureUtils.exception(new MetadataStoreException.BadVersionException(""));
109-
} else {
110-
long newVersion = existingValue != null ? existingValue.getVersion() + 1 : 0;
111-
long createdTimestamp =
112-
existingValue != null ? existingValue.getCreatedTimestamp() : now;
113-
Value newValue = new Value(newVersion, data, createdTimestamp, now,
114-
options.contains(CreateOption.Ephemeral));
115-
persistedData.put(path, data);
116-
map.put(path, newValue);
117-
118-
NotificationType type =
119-
existingValue == null ? NotificationType.Created :
120-
NotificationType.Modified;
121-
receivedNotification(new Notification(type, path));
122-
if (type == NotificationType.Created) {
123-
notifyParentChildrenChanged(path);
124-
}
125-
return FutureUtils
126-
.value(new Stat(path, newValue.getVersion(), newValue.getCreatedTimestamp(),
127-
newValue.getModifiedTimestamp(),
128-
false, true));
129-
}
130-
}
131-
}
68+
metadataStore = new FaultInjectionMetadataStore(
69+
MetadataStoreExtended.create("memory:local", MetadataStoreConfig.builder().build())) {
70+
@Override
71+
public CompletableFuture<Stat> put(String path, byte[] value, Optional<Long> expectedVersion) {
72+
CompletableFuture<Stat> future = super.put(path, value, expectedVersion);
73+
future.whenComplete((res, e) -> {
74+
if (e != null) {
75+
return;
13276
}
133-
};
134-
metadataStore = new FaultInjectionMetadataStore(metadataStoreExtended);
77+
persistedData.put(path, value);
78+
});
79+
return future;
80+
}
81+
};
13582
bkc = new PulsarMockBookKeeper(executor) {
13683
@Override
13784
public void asyncDeleteLedger(long lId, AsyncCallback.DeleteCallback cb, Object ctx) {

‎pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/LocalMemoryMetadataStore.java

+3-5
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import java.util.concurrent.CompletableFuture;
3333
import java.util.concurrent.atomic.AtomicLong;
3434
import lombok.Data;
35-
import lombok.Getter;
3635
import lombok.extern.slf4j.Slf4j;
3736
import org.apache.bookkeeper.common.concurrent.FutureUtils;
3837
import org.apache.pulsar.common.util.FutureUtil;
@@ -53,17 +52,16 @@ public class LocalMemoryMetadataStore extends AbstractMetadataStore implements M
5352
static final String MEMORY_SCHEME_IDENTIFIER = "memory:";
5453

5554
@Data
56-
@Getter
57-
protected static class Value {
55+
private static class Value {
5856
final long version;
5957
final byte[] data;
6058
final long createdTimestamp;
6159
final long modifiedTimestamp;
6260
final boolean ephemeral;
6361
}
6462

65-
protected final NavigableMap<String, Value> map;
66-
protected final AtomicLong sequentialIdGenerator;
63+
private final NavigableMap<String, Value> map;
64+
private final AtomicLong sequentialIdGenerator;
6765

6866
private static final Map<String, NavigableMap<String, Value>> STATIC_MAPS = new MapMaker()
6967
.weakValues().makeMap();

0 commit comments

Comments
 (0)