eolivelli commented on a change in pull request #2936:
URL: https://github.com/apache/bookkeeper/pull/2936#discussion_r771811531
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java
##########
@@ -327,6 +328,13 @@ public boolean ledgerExists(long ledgerId) throws
IOException {
return ledgerCache.ledgerExists(ledgerId);
}
+ @Override
+ public boolean entryExists(long ledgerId, long entryId) throws IOException
{
+ //Implementation should be as simple as what's below, but this needs
testing
+ //return ledgerCache.getEntryOffset(ledgerId, entryId) > 0;
+ throw new UnsupportedOperationException("entry exists not supported");
Review comment:
is it better to throw a IOException (here and in the other methods in
this class) ?
Unchecked exception are usually not handled and so we can fall into
unexpected errors.
not a big deal in any case, as the bookie is not working
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/datainteg/DataIntegrityCookieValidation.java
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.bookie.datainteg;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.UnknownHostException;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.bookie.BookieImpl;
+import org.apache.bookkeeper.bookie.Cookie;
+import org.apache.bookkeeper.bookie.CookieValidation;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An implementation of the CookieValidation interface that allows for
auto-stamping
+ * cookies when configured and used in conjunction with the data integrity
service.
+ * Because the data integrity service can heal a bookie with lost data due to
a disk
+ * failure, a bookie can auto stamp new cookies as part of the healing process.
+ */
+public class DataIntegrityCookieValidation implements CookieValidation {
+ private static final Logger log =
LoggerFactory.getLogger(DataIntegrityCookieValidation.class);
+ private final ServerConfiguration conf;
+ private final BookieId bookieId;
+ private final RegistrationManager registrationManager;
+ private final DataIntegrityCheck dataIntegCheck;
+
+ public DataIntegrityCookieValidation(ServerConfiguration conf,
+ RegistrationManager
registrationManager,
+ DataIntegrityCheck dataIntegCheck)
+ throws UnknownHostException {
+ this.conf = conf;
+ this.registrationManager = registrationManager;
+ this.bookieId = BookieImpl.getBookieId(conf);
+ this.dataIntegCheck = dataIntegCheck;
+ }
+
+ private Optional<Versioned<Cookie>> getRegManagerCookie() throws
BookieException {
+ try {
+ return
Optional.of(Cookie.readFromRegistrationManager(registrationManager, bookieId));
+ } catch (BookieException.CookieNotFoundException noCookieException) {
+ return Optional.empty();
+ }
+ }
+
+ private List<Optional<Cookie>> collectDirectoryCookies(List<File>
directories) throws BookieException {
+ List<Optional<Cookie>> cookies = new ArrayList<>();
+ for (File d : directories) {
+ try {
+ cookies.add(Optional.of(Cookie.readFromDirectory(d)));
+ } catch (FileNotFoundException fnfe) {
+ cookies.add(Optional.empty());
+ } catch (IOException ioe) {
+ throw new BookieException.InvalidCookieException(ioe);
+ }
+ }
+ return cookies;
+ }
+
+ private void stampCookie(Cookie masterCookie, Version expectedVersion,
List<File> directories)
+ throws BookieException {
+ // stamp to ZK first as it's the authoritive cookie. If this fails
part way through
+ // stamping the directories, then a data integrity check will occur.
+ log.info("Stamping cookie to ZK");
+ masterCookie.writeToRegistrationManager(registrationManager, conf,
expectedVersion);
+ for (File d : directories) {
+ try {
+ log.info("Stamping cookie to directory {}", d);
+ masterCookie.writeToDirectory(d);
+ } catch (IOException ioe) {
+ log.error("Exception writing cookie to {}", ioe);
+ throw new BookieException.InvalidCookieException(ioe);
+ }
+ }
+ }
+
+ @Override
+ public void checkCookies(List<File> directories)
+ throws BookieException, InterruptedException {
+ String instanceId = registrationManager.getClusterInstanceId();
+ if (instanceId == null) {
+ throw new BookieException.InvalidCookieException("Cluster instance
ID unavailable");
+ }
+ Cookie masterCookie;
+ try {
+ masterCookie =
Cookie.generateCookie(conf).setInstanceId(instanceId).build();
+ } catch (UnknownHostException uhe) {
+ throw new BookieException.InvalidCookieException(uhe);
+ }
+
+ // collect existing cookies
+ Optional<Versioned<Cookie>> regManagerCookie = getRegManagerCookie();
+ List<Optional<Cookie>> directoryCookies =
collectDirectoryCookies(directories);
+
+ // if master is empty, everything must be empty, otherwise the cluster
is messed up
+ if (!regManagerCookie.isPresent()) {
+ // if everything is empty, it's a new install, just stamp the
cookies
+ if (directoryCookies.stream().noneMatch(Optional::isPresent)) {
+ log.info("New environment found. Stamping cookies");
+ stampCookie(masterCookie, Version.NEW, directories);
+ } else {
+ String errorMsg =
+ "Cookie missing from ZK. Either it was manually deleted, "
+ + "or the bookie was started pointing to a different ZK
cluster "
+ + "than the one it was originally started with. "
+ + "This requires manual intervention to fix";
+ log.error(errorMsg);
+ throw new BookieException.InvalidCookieException(errorMsg);
+ }
+ } else if (!regManagerCookie.get().getValue().equals(masterCookie)
+ || !directoryCookies.stream().allMatch(c ->
c.map(masterCookie::equals).orElse(false))) {
+ if (conf.isDataIntegrityStampMissingCookiesEnabled()) {
+ log.warn("ZK cookie({}) or directory cookies({}) do not match
master cookie ({}), running check",
+ regManagerCookie, directoryCookies, masterCookie);
+ try {
+ dataIntegCheck.runPreBoot("INVALID_COOKIE").get();
+ } catch (ExecutionException ee) {
+ if (ee.getCause() instanceof BookieException) {
+ throw (BookieException) ee.getCause();
+ } else {
+ throw new
BookieException.InvalidCookieException(ee.getCause());
+ }
+ }
+ log.info("Environment should be in a sane state. Stamp new
cookies");
+ stampCookie(masterCookie, regManagerCookie.get().getVersion(),
directories);
+ } else {
+ String errorMsg = MessageFormat.format(
+ "ZK cookie({0}) or directory cookies({1}) do not match
master cookie ({2})"
+ + " and missing cookie stamping is disabled.",
+ regManagerCookie, directoryCookies, masterCookie);
+ log.error(errorMsg);
+ throw new BookieException.InvalidCookieException(errorMsg);
+ }
+ } // else all cookies match the masterCookie, meaning nothing has
changed in the configuration
Review comment:
what about logging something ?
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/UncleanShutdownDetectionImpl.java
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.bookie;
+
+import java.io.File;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Used to determine if the prior shutdown was unclean or not. It does so
+ * by adding a file to each ledger directory after successful start-up
+ * and removing the file on graceful shutdown.
+ * Any abrupt termination will cause one or more of these files to not be
cleared
+ * and so on the subsequent boot-up, the presence of any of these files will
+ * indicate an unclean shutdown.
+ */
+public class UncleanShutdownDetectionImpl implements UncleanShutdownDetection {
+ private static final Logger LOG =
LoggerFactory.getLogger(UncleanShutdownDetectionImpl.class);
+ private final LedgerDirsManager ledgerDirsManager;
+ static final String DirtyFileName = "DIRTY";
Review comment:
nit: DIRTY_FILENAME is a better name
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/UncleanShutdownDetectionImpl.java
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.bookie;
+
+import java.io.File;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Used to determine if the prior shutdown was unclean or not. It does so
+ * by adding a file to each ledger directory after successful start-up
+ * and removing the file on graceful shutdown.
+ * Any abrupt termination will cause one or more of these files to not be
cleared
+ * and so on the subsequent boot-up, the presence of any of these files will
+ * indicate an unclean shutdown.
+ */
+public class UncleanShutdownDetectionImpl implements UncleanShutdownDetection {
+ private static final Logger LOG =
LoggerFactory.getLogger(UncleanShutdownDetectionImpl.class);
+ private final LedgerDirsManager ledgerDirsManager;
+ static final String DirtyFileName = "DIRTY";
+
+ public UncleanShutdownDetectionImpl(LedgerDirsManager ledgerDirsManager) {
+ this.ledgerDirsManager = ledgerDirsManager;
+ }
+
+ @Override
+ public void registerStartUp() {
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ try {
+ File dirtyFile = new File(ledgerDir, DirtyFileName);
+ dirtyFile.createNewFile();
+ } catch (Throwable t) {
+ LOG.error("Unable to register start-up (so an unclean shutdown
cannot"
Review comment:
+1 for crashing in this case
can we also log the we created this file ?
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/UncleanShutdownDetectionImpl.java
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.bookie;
+
+import java.io.File;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Used to determine if the prior shutdown was unclean or not. It does so
+ * by adding a file to each ledger directory after successful start-up
+ * and removing the file on graceful shutdown.
+ * Any abrupt termination will cause one or more of these files to not be
cleared
+ * and so on the subsequent boot-up, the presence of any of these files will
+ * indicate an unclean shutdown.
+ */
+public class UncleanShutdownDetectionImpl implements UncleanShutdownDetection {
+ private static final Logger LOG =
LoggerFactory.getLogger(UncleanShutdownDetectionImpl.class);
+ private final LedgerDirsManager ledgerDirsManager;
+ static final String DirtyFileName = "DIRTY";
+
+ public UncleanShutdownDetectionImpl(LedgerDirsManager ledgerDirsManager) {
+ this.ledgerDirsManager = ledgerDirsManager;
+ }
+
+ @Override
+ public void registerStartUp() {
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ try {
+ File dirtyFile = new File(ledgerDir, DirtyFileName);
+ dirtyFile.createNewFile();
+ } catch (Throwable t) {
+ LOG.error("Unable to register start-up (so an unclean shutdown
cannot"
+ + " be detected). Dirty file of ledger dir {} could
not be created",
+ ledgerDir.getAbsolutePath(), t);
+ }
+ }
+ }
+
+ @Override
+ public void registerCleanShutdown() {
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ try {
+ File dirtyFile = new File(ledgerDir, DirtyFileName);
+ dirtyFile.delete();
Review comment:
we should check the return value here (I wonder why spotbugs didn't
complaint)
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/datainteg/DataIntegrityCheck.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.bookie.datainteg;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * The interface for the data integrity check feature. This feature allows
+ * a bookie to handle data loss scenarios such as when running without
+ * the journal or after a disk failure has caused the loss of all data.
+ */
+public interface DataIntegrityCheck {
+ /**
+ * Run quick preboot check. This check should do enough to ensure that
+ * it is safe to complete the boot sequence without compromising
correctness.
+ * To this end, if it finds that this bookie is part of the last ensemble
of
+ * an unclosed ledger, it must prevent the bookie from being able store new
+ * entries for that ledger and must prevent the bookie from taking part in
+ * the discovery of the last entry of that ledger.
+ */
+ CompletableFuture<Void> runPreBoot(String reason);
+
+ /**
+ * Whether we need to run a full check.
+ * This condition can be set by the runPreBoot() call to run a full check
+ * in the background once the bookie is running. This can later be used
+ * to run the full check periodically, or to exponentially backoff and
retry
+ * when some transient condition prevents a ledger being fixed during a
+ * full check.
+ */
+ boolean needsFull() throws IOException;
Review comment:
nit: needsFullCheck
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
##########
@@ -2288,6 +2288,8 @@ private int statusCodeToExceptionCode(StatusCode status) {
return BKException.Code.WriteOnReadOnlyBookieException;
case ETOOMANYREQUESTS:
return BKException.Code.TooManyRequestsException;
+ case EUNKNOWN:
Review comment:
what about naming this EUNKNOWNLEDGERSTATE ? otherwise people can think
that is is kind of "unknown/generic error"
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/datainteg/DataIntegrityCheckImpl.java
##########
@@ -0,0 +1,554 @@
+/*
+ * 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.bookie.datainteg;
+
+import com.google.common.collect.ImmutableSortedMap;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Scheduler;
+import io.reactivex.rxjava3.core.Single;
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.bookie.LedgerStorage;
+import org.apache.bookkeeper.bookie.LedgerStorage.StorageState;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.meta.LedgerManager;
+import org.apache.bookkeeper.net.BookieId;
+
+/**
+ * An implementation of the DataIntegrityCheck interface.
+ */
+@Slf4j
+public class DataIntegrityCheckImpl implements DataIntegrityCheck {
+ private static final int MAX_INFLIGHT = 300;
+ private static final int MAX_ENTRIES_INFLIGHT = 3000;
+ private static final int ZK_TIMEOUT_S = 30;
+ private final BookieId bookieId;
+ private final LedgerManager ledgerManager;
+ private final LedgerStorage ledgerStorage;
+ private final EntryCopier entryCopier;
+ private final BookKeeperAdmin admin;
+ private final Scheduler scheduler;
+ private final AtomicReference<Map<Long, LedgerMetadata>> ledgersCacheRef =
+ new AtomicReference<>(null);
+ private CompletableFuture<Void> preBootFuture;
+
+ public DataIntegrityCheckImpl(BookieId bookieId,
+ LedgerManager ledgerManager,
+ LedgerStorage ledgerStorage,
+ EntryCopier entryCopier,
+ BookKeeperAdmin admin,
+ Scheduler scheduler) {
+ this.bookieId = bookieId;
+ this.ledgerManager = ledgerManager;
+ this.ledgerStorage = ledgerStorage;
+ this.entryCopier = entryCopier;
+ this.admin = admin;
+ this.scheduler = scheduler;
+ }
+
+ @Override
+ public CompletableFuture<Void> runPreBoot(String reason) {
+ // we only run this once, it could be kicked off by different checks
+ synchronized (this) {
+ if (preBootFuture == null) {
+ preBootFuture = runPreBootSequence(reason);
+ }
+ }
+ return preBootFuture;
+
+ }
+
+ private CompletableFuture<Void> runPreBootSequence(String reason) {
+ String runId = UUID.randomUUID().toString();
+ log.info("Event: {}, RunId: {}, Reason: {}", Events.PREBOOT_START,
runId, reason);
+ try {
+
this.ledgerStorage.setStorageStateFlag(StorageState.NEEDS_INTEGRITY_CHECK);
+ } catch (IOException ioe) {
+ log.error("Event: {}, RunId: {}", Events.PREBOOT_ERROR, runId,
ioe);
+ return FutureUtils.exception(ioe);
+ }
+
+ MetadataAsyncIterator iter = new MetadataAsyncIterator(scheduler,
+ ledgerManager, MAX_INFLIGHT, ZK_TIMEOUT_S, TimeUnit.SECONDS);
+ CompletableFuture<Void> promise = new CompletableFuture<>();
+ Map<Long, LedgerMetadata> ledgersCache =
Review comment:
we can keep this here, and address the problem in #1949
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/datainteg/DataIntegrityCheckImpl.java
##########
@@ -0,0 +1,554 @@
+/*
+ * 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.bookie.datainteg;
+
+import com.google.common.collect.ImmutableSortedMap;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Scheduler;
+import io.reactivex.rxjava3.core.Single;
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.bookie.LedgerStorage;
+import org.apache.bookkeeper.bookie.LedgerStorage.StorageState;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.meta.LedgerManager;
+import org.apache.bookkeeper.net.BookieId;
+
+/**
+ * An implementation of the DataIntegrityCheck interface.
+ */
+@Slf4j
+public class DataIntegrityCheckImpl implements DataIntegrityCheck {
+ private static final int MAX_INFLIGHT = 300;
+ private static final int MAX_ENTRIES_INFLIGHT = 3000;
+ private static final int ZK_TIMEOUT_S = 30;
+ private final BookieId bookieId;
+ private final LedgerManager ledgerManager;
+ private final LedgerStorage ledgerStorage;
+ private final EntryCopier entryCopier;
+ private final BookKeeperAdmin admin;
+ private final Scheduler scheduler;
+ private final AtomicReference<Map<Long, LedgerMetadata>> ledgersCacheRef =
+ new AtomicReference<>(null);
+ private CompletableFuture<Void> preBootFuture;
+
+ public DataIntegrityCheckImpl(BookieId bookieId,
+ LedgerManager ledgerManager,
+ LedgerStorage ledgerStorage,
+ EntryCopier entryCopier,
+ BookKeeperAdmin admin,
+ Scheduler scheduler) {
+ this.bookieId = bookieId;
+ this.ledgerManager = ledgerManager;
+ this.ledgerStorage = ledgerStorage;
+ this.entryCopier = entryCopier;
+ this.admin = admin;
+ this.scheduler = scheduler;
+ }
+
+ @Override
+ public CompletableFuture<Void> runPreBoot(String reason) {
+ // we only run this once, it could be kicked off by different checks
+ synchronized (this) {
+ if (preBootFuture == null) {
+ preBootFuture = runPreBootSequence(reason);
+ }
+ }
+ return preBootFuture;
+
+ }
+
+ private CompletableFuture<Void> runPreBootSequence(String reason) {
+ String runId = UUID.randomUUID().toString();
+ log.info("Event: {}, RunId: {}, Reason: {}", Events.PREBOOT_START,
runId, reason);
+ try {
+
this.ledgerStorage.setStorageStateFlag(StorageState.NEEDS_INTEGRITY_CHECK);
+ } catch (IOException ioe) {
+ log.error("Event: {}, RunId: {}", Events.PREBOOT_ERROR, runId,
ioe);
+ return FutureUtils.exception(ioe);
+ }
+
+ MetadataAsyncIterator iter = new MetadataAsyncIterator(scheduler,
+ ledgerManager, MAX_INFLIGHT, ZK_TIMEOUT_S, TimeUnit.SECONDS);
+ CompletableFuture<Void> promise = new CompletableFuture<>();
+ Map<Long, LedgerMetadata> ledgersCache =
+ new
ConcurrentSkipListMap<>(Comparator.<Long>naturalOrder().reversed());
+ iter.forEach((ledgerId, metadata) -> {
+ if (ensemblesContainBookie(metadata, bookieId)) {
+ ledgersCache.put(ledgerId, metadata);
+ try {
+ if (!ledgerStorage.ledgerExists(ledgerId)) {
+ ledgerStorage.setMasterKey(ledgerId, new byte[0]);
+ }
+ } catch (IOException ioe) {
+ return FutureUtils.exception(ioe);
Review comment:
what about logging ledgerId...or other useful context ?
##########
File path:
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/UncleanShutdownDetectionImpl.java
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.bookie;
+
+import java.io.File;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Used to determine if the prior shutdown was unclean or not. It does so
+ * by adding a file to each ledger directory after successful start-up
+ * and removing the file on graceful shutdown.
+ * Any abrupt termination will cause one or more of these files to not be
cleared
+ * and so on the subsequent boot-up, the presence of any of these files will
+ * indicate an unclean shutdown.
+ */
+public class UncleanShutdownDetectionImpl implements UncleanShutdownDetection {
+ private static final Logger LOG =
LoggerFactory.getLogger(UncleanShutdownDetectionImpl.class);
+ private final LedgerDirsManager ledgerDirsManager;
+ static final String DirtyFileName = "DIRTY";
+
+ public UncleanShutdownDetectionImpl(LedgerDirsManager ledgerDirsManager) {
+ this.ledgerDirsManager = ledgerDirsManager;
+ }
+
+ @Override
+ public void registerStartUp() {
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ try {
+ File dirtyFile = new File(ledgerDir, DirtyFileName);
+ dirtyFile.createNewFile();
+ } catch (Throwable t) {
+ LOG.error("Unable to register start-up (so an unclean shutdown
cannot"
+ + " be detected). Dirty file of ledger dir {} could
not be created",
+ ledgerDir.getAbsolutePath(), t);
+ }
+ }
+ }
+
+ @Override
+ public void registerCleanShutdown() {
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ try {
+ File dirtyFile = new File(ledgerDir, DirtyFileName);
+ dirtyFile.delete();
+ } catch (Throwable t) {
+ LOG.error("Unable to register a clean shutdown, dirty file of "
+ + " ledger dir {} could not be deleted",
+ ledgerDir.getAbsolutePath(), t);
+ }
+ }
+ }
+
+ @Override
+ public boolean lastShutdownWasUnclean() {
+ try {
+ for (File ledgerDir : ledgerDirsManager.getAllLedgerDirs()) {
+ File dirtyFile = new File(ledgerDir, DirtyFileName);
+ if (dirtyFile.exists()) {
+ return true;
Review comment:
can we log something, like "Detected un clean shutdown as file xxx
exists"
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
