You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2023/10/30 23:27:47 UTC
(kafka) branch trunk updated: KAFKA-15582: Move the clean shutdown file to the storage package (#14603)
This is an automated email from the ASF dual-hosted git repository.
junrao pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 8f8ad6db384 KAFKA-15582: Move the clean shutdown file to the storage package (#14603)
8f8ad6db384 is described below
commit 8f8ad6db384ce30e8a5d848d3ed826a3f7a54dfe
Author: Calvin Liu <83...@users.noreply.github.com>
AuthorDate: Mon Oct 30 16:27:40 2023 -0700
KAFKA-15582: Move the clean shutdown file to the storage package (#14603)
A follow-up change to move the clean shutdown file to the storage package.
Reviewers: Ismael Juma <is...@juma.me.uk>, Jun Rao <ju...@gmail.com>
---
checkstyle/import-control-core.xml | 21 +++++-----
checkstyle/import-control-storage.xml | 1 +
core/src/main/scala/kafka/log/LogManager.scala | 19 ++++-----
.../kafka/server/BrokerLifecycleManager.scala | 10 +++--
core/src/main/scala/kafka/server/KafkaServer.scala | 3 +-
.../test/scala/unit/kafka/log/LogLoaderTest.scala | 1 +
.../test/scala/unit/kafka/log/LogManagerTest.scala | 7 ++--
.../kafka/server/BrokerLifecycleManagerTest.scala | 10 ++---
...chDrivenReplicationProtocolAcceptanceTest.scala | 2 +-
.../checkpoint}/CleanShutdownFileHandler.java | 46 +++++++++++-----------
.../checkpoint}/CleanShutdownFileHandlerTest.java | 30 ++++++++++++--
11 files changed, 89 insertions(+), 61 deletions(-)
diff --git a/checkstyle/import-control-core.xml b/checkstyle/import-control-core.xml
index 983f077adc1..849c45e5b19 100644
--- a/checkstyle/import-control-core.xml
+++ b/checkstyle/import-control-core.xml
@@ -73,19 +73,16 @@
<allow pkg="org.apache.kafka.clients" />
</subpackage>
- <subpackage name="log">
- <allow pkg="com.fasterxml.jackson" />
+ <subpackage name="log.remote">
+ <allow pkg="org.apache.kafka.server.common" />
+ <allow pkg="org.apache.kafka.server.log.remote" />
+ <allow pkg="org.apache.kafka.server.metrics" />
+ <allow pkg="org.apache.kafka.storage.internals" />
+ <allow pkg="kafka.log" />
+ <allow pkg="kafka.cluster" />
+ <allow pkg="kafka.server" />
+ <allow pkg="org.mockito" />
<allow pkg="org.apache.kafka.test" />
- <subpackage name="remote">
- <allow pkg="org.apache.kafka.server.common" />
- <allow pkg="org.apache.kafka.server.log.remote" />
- <allow pkg="org.apache.kafka.server.metrics" />
- <allow pkg="org.apache.kafka.storage.internals" />
- <allow pkg="kafka.log" />
- <allow pkg="kafka.cluster" />
- <allow pkg="kafka.server" />
- <allow pkg="org.mockito" />
- </subpackage>
</subpackage>
<subpackage name="server">
diff --git a/checkstyle/import-control-storage.xml b/checkstyle/import-control-storage.xml
index 2e0b85dcf76..42d170ec315 100644
--- a/checkstyle/import-control-storage.xml
+++ b/checkstyle/import-control-storage.xml
@@ -75,6 +75,7 @@
</subpackage>
<subpackage name="storage.internals">
+ <allow pkg="com.fasterxml.jackson" />
<allow pkg="com.yammer.metrics.core" />
<allow pkg="org.apache.kafka.server"/>
<allow pkg="org.apache.kafka.storage.internals"/>
diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala
index 61fe4168cd3..c7c7ffd2336 100755
--- a/core/src/main/scala/kafka/log/LogManager.scala
+++ b/core/src/main/scala/kafka/log/LogManager.scala
@@ -36,12 +36,13 @@ import scala.util.{Failure, Success, Try}
import kafka.utils.Implicits._
import org.apache.kafka.common.config.TopicConfig
-import java.util.Properties
+import java.util.{OptionalLong, Properties}
import org.apache.kafka.server.common.MetadataVersion
import org.apache.kafka.storage.internals.log.LogConfig.MessageFormatVersion
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.Scheduler
import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig, RemoteIndexCache}
+import org.apache.kafka.storage.internals.checkpoint.CleanShutdownFileHandler
import scala.annotation.nowarn
@@ -1430,27 +1431,27 @@ class LogManager(logDirs: Seq[File],
}
}
- def readBrokerEpochFromCleanShutdownFiles(): Long = {
+ def readBrokerEpochFromCleanShutdownFiles(): OptionalLong = {
// Verify whether all the log dirs have the same broker epoch in their clean shutdown files. If there is any dir not
// live, fail the broker epoch check.
if (liveLogDirs.size < logDirs.size) {
- return -1L
+ return OptionalLong.empty()
}
var brokerEpoch = -1L
for (dir <- liveLogDirs) {
val cleanShutdownFileHandler = new CleanShutdownFileHandler(dir.getPath)
val currentBrokerEpoch = cleanShutdownFileHandler.read
- if (currentBrokerEpoch == -1L) {
+ if (!currentBrokerEpoch.isPresent) {
info(s"Unable to read the broker epoch in ${dir.toString}.")
- return -1L
+ return OptionalLong.empty()
}
- if (brokerEpoch != -1 && currentBrokerEpoch != brokerEpoch) {
+ if (brokerEpoch != -1 && currentBrokerEpoch.getAsLong != brokerEpoch) {
info(s"Found different broker epochs in ${dir.toString}. Other=$brokerEpoch vs current=$currentBrokerEpoch.")
- return -1L
+ return OptionalLong.empty()
}
- brokerEpoch = currentBrokerEpoch
+ brokerEpoch = currentBrokerEpoch.getAsLong
}
- brokerEpoch
+ OptionalLong.of(brokerEpoch)
}
}
diff --git a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
index 112c6f57987..36d0ee43873 100644
--- a/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
+++ b/core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
@@ -30,6 +30,8 @@ import org.apache.kafka.metadata.{BrokerState, VersionRange}
import org.apache.kafka.queue.EventQueue.DeadlineFunction
import org.apache.kafka.common.utils.{ExponentialBackoff, LogContext, Time}
import org.apache.kafka.queue.{EventQueue, KafkaEventQueue}
+
+import java.util.OptionalLong
import scala.jdk.CollectionConverters._
@@ -188,9 +190,9 @@ class BrokerLifecycleManager(
private var _channelManager: NodeToControllerChannelManager = _
/**
- * The broker epoch from the previous run, or -1 if the epoch is not able to be found.
+ * The broker epoch from the previous run, or empty if the epoch is not found.
*/
- @volatile private var previousBrokerEpoch: Long = -1L
+ @volatile private var previousBrokerEpoch: OptionalLong = OptionalLong.empty()
/**
* The event queue.
@@ -216,7 +218,7 @@ class BrokerLifecycleManager(
clusterId: String,
advertisedListeners: ListenerCollection,
supportedFeatures: util.Map[String, VersionRange],
- previousBrokerEpoch: Long): Unit = {
+ previousBrokerEpoch: OptionalLong): Unit = {
this.previousBrokerEpoch = previousBrokerEpoch
eventQueue.append(new StartupEvent(highestMetadataOffsetProvider,
channelManager, clusterId, advertisedListeners, supportedFeatures))
@@ -322,7 +324,7 @@ class BrokerLifecycleManager(
setIncarnationId(incarnationId).
setListeners(_advertisedListeners).
setRack(rack.orNull).
- setPreviousBrokerEpoch(previousBrokerEpoch)
+ setPreviousBrokerEpoch(previousBrokerEpoch.orElse(-1L))
if (isDebugEnabled) {
debug(s"Sending broker registration $data")
}
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 038df14cc91..b6cea90abea 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -61,6 +61,7 @@ import org.apache.zookeeper.client.ZKClientConfig
import java.io.{File, IOException}
import java.net.{InetAddress, SocketTimeoutException}
+import java.util.OptionalLong
import java.util.concurrent._
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import scala.collection.{Map, Seq}
@@ -438,7 +439,7 @@ class KafkaServer(
kraftMetaProps.clusterId,
networkListeners,
ibpAsFeature,
- -1
+ OptionalLong.empty()
)
logger.debug("Start RaftManager")
}
diff --git a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
index da19c7055ac..ac18c07dced 100644
--- a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
@@ -34,6 +34,7 @@ import org.apache.kafka.server.common.MetadataVersion.IBP_0_11_0_IV0
import org.apache.kafka.server.util.{MockTime, Scheduler}
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
import org.apache.kafka.storage.internals.log.{AbortedTxn, CleanerConfig, EpochEntry, FetchDataInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogSegment, LogSegments, LogStartOffsetIncrementReason, OffsetIndex, ProducerStateManager, ProducerStateManagerConfig, SnapshotFile}
+import org.apache.kafka.storage.internals.checkpoint.CleanShutdownFileHandler
import org.junit.jupiter.api.Assertions.{assertDoesNotThrow, assertEquals, assertFalse, assertNotEquals, assertThrows, assertTrue}
import org.junit.jupiter.api.function.Executable
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
index eda800abb11..b0dd81e1c58 100755
--- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
@@ -36,10 +36,11 @@ import org.mockito.Mockito.{doAnswer, doNothing, mock, never, spy, times, verify
import java.io._
import java.nio.file.Files
import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap, Future}
-import java.util.{Collections, Properties}
+import java.util.{Collections, OptionalLong, Properties}
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.storage.internals.log.{FetchDataInfo, FetchIsolation, LogConfig, LogDirFailureChannel, ProducerStateManagerConfig, RemoteIndexCache}
+import org.apache.kafka.storage.internals.checkpoint.CleanShutdownFileHandler
import scala.collection.{Map, mutable}
import scala.collection.mutable.ArrayBuffer
@@ -131,7 +132,7 @@ class LogManagerTest {
assertFalse(Files.exists(new File(logDir1, CleanShutdownFileHandler.CLEAN_SHUTDOWN_FILE_NAME).toPath))
assertTrue(Files.exists(new File(logDir2, CleanShutdownFileHandler.CLEAN_SHUTDOWN_FILE_NAME).toPath))
- assertEquals(-1L, logManagerForTest.get.readBrokerEpochFromCleanShutdownFiles())
+ assertEquals(OptionalLong.empty(), logManagerForTest.get.readBrokerEpochFromCleanShutdownFiles())
} finally {
logManagerForTest.foreach(manager => manager.liveLogDirs.foreach(Utils.delete))
}
@@ -159,7 +160,7 @@ class LogManagerTest {
assertTrue(Files.exists(new File(logDir1, CleanShutdownFileHandler.CLEAN_SHUTDOWN_FILE_NAME).toPath))
assertTrue(Files.exists(new File(logDir2, CleanShutdownFileHandler.CLEAN_SHUTDOWN_FILE_NAME).toPath))
- assertEquals(3L, logManagerForTest.get.readBrokerEpochFromCleanShutdownFiles())
+ assertEquals(OptionalLong.of(3L), logManagerForTest.get.readBrokerEpochFromCleanShutdownFiles())
} finally {
logManagerForTest.foreach(manager => manager.liveLogDirs.foreach(Utils.delete))
}
diff --git a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
index 90d93466c29..77ae31c4c2e 100644
--- a/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala
@@ -17,7 +17,7 @@
package kafka.server
-import java.util.{Collections, Properties}
+import java.util.{Collections, OptionalLong, Properties}
import kafka.utils.TestUtils
import org.apache.kafka.common.Node
import org.apache.kafka.common.message.{BrokerHeartbeatResponseData, BrokerRegistrationResponseData}
@@ -55,7 +55,7 @@ class BrokerLifecycleManagerTest {
assertEquals(BrokerState.NOT_RUNNING, manager.state)
manager.start(() => context.highestMetadataOffset.get(),
context.mockChannelManager, context.clusterId, context.advertisedListeners,
- Collections.emptyMap(), -1)
+ Collections.emptyMap(), OptionalLong.empty())
TestUtils.retry(60000) {
assertEquals(BrokerState.STARTING, manager.state)
}
@@ -71,7 +71,7 @@ class BrokerLifecycleManagerTest {
context.controllerNodeProvider.node.set(controllerNode)
manager.start(() => context.highestMetadataOffset.get(),
context.mockChannelManager, context.clusterId, context.advertisedListeners,
- Collections.emptyMap(), 10L)
+ Collections.emptyMap(), OptionalLong.of(10L))
TestUtils.retry(60000) {
assertEquals(1, context.mockChannelManager.unsentQueue.size)
assertEquals(10L, context.mockChannelManager.unsentQueue.getFirst.request.build().asInstanceOf[BrokerRegistrationRequest].data().previousBrokerEpoch())
@@ -101,7 +101,7 @@ class BrokerLifecycleManagerTest {
assertEquals(1, context.mockClient.futureResponses().size)
manager.start(() => context.highestMetadataOffset.get(),
context.mockChannelManager, context.clusterId, context.advertisedListeners,
- Collections.emptyMap(), -1)
+ Collections.emptyMap(), OptionalLong.empty())
// We should send the first registration request and get a failure immediately
TestUtils.retry(60000) {
context.poll()
@@ -139,7 +139,7 @@ class BrokerLifecycleManagerTest {
new BrokerHeartbeatResponseData().setIsCaughtUp(true)), controllerNode)
manager.start(() => context.highestMetadataOffset.get(),
context.mockChannelManager, context.clusterId, context.advertisedListeners,
- Collections.emptyMap(), -1)
+ Collections.emptyMap(), OptionalLong.empty())
TestUtils.retry(10000) {
context.poll()
manager.eventQueue.wakeup()
diff --git a/core/src/test/scala/unit/kafka/server/epoch/EpochDrivenReplicationProtocolAcceptanceTest.scala b/core/src/test/scala/unit/kafka/server/epoch/EpochDrivenReplicationProtocolAcceptanceTest.scala
index f8571316f0f..e24f348ad68 100644
--- a/core/src/test/scala/unit/kafka/server/epoch/EpochDrivenReplicationProtocolAcceptanceTest.scala
+++ b/core/src/test/scala/unit/kafka/server/epoch/EpochDrivenReplicationProtocolAcceptanceTest.scala
@@ -17,7 +17,6 @@
package kafka.server.epoch
-import kafka.log.CleanShutdownFileHandler
import kafka.log.UnifiedLog
import kafka.server.KafkaConfig._
import kafka.server.{KafkaConfig, KafkaServer, QuorumTestHarness}
@@ -32,6 +31,7 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer
import org.apache.kafka.server.common.MetadataVersion
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
import org.apache.kafka.storage.internals.log.EpochEntry
+import org.apache.kafka.storage.internals.checkpoint.CleanShutdownFileHandler
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo}
diff --git a/core/src/main/java/kafka/log/CleanShutdownFileHandler.java b/storage/src/main/java/org/apache/kafka/storage/internals/checkpoint/CleanShutdownFileHandler.java
similarity index 75%
rename from core/src/main/java/kafka/log/CleanShutdownFileHandler.java
rename to storage/src/main/java/org/apache/kafka/storage/internals/checkpoint/CleanShutdownFileHandler.java
index a14cc6a1b66..321b46ce9d5 100644
--- a/core/src/main/java/kafka/log/CleanShutdownFileHandler.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/checkpoint/CleanShutdownFileHandler.java
@@ -15,11 +15,12 @@
* limitations under the License.
*/
-package kafka.log;
+package org.apache.kafka.storage.internals.checkpoint;
-import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.Json;
import org.slf4j.Logger;
import java.io.File;
@@ -28,9 +29,7 @@ import java.io.BufferedWriter;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
-import java.util.HashMap;
-import java.util.Locale;
-import java.util.Map;
+import java.util.OptionalLong;
/**
* Clean shutdown file that indicates the broker was cleanly shutdown in 0.8 and higher.
@@ -45,17 +44,21 @@ import java.util.Map;
public class CleanShutdownFileHandler {
public static final String CLEAN_SHUTDOWN_FILE_NAME = ".kafka_cleanshutdown";
- private final File cleanShutdownFile;
+ // Visible for testing
+ final File cleanShutdownFile;
private static final int CURRENT_VERSION = 0;
private final Logger logger;
- private enum Fields {
- VERSION,
- BROKER_EPOCH;
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ private static class Content {
+ public int version;
+ public Long brokerEpoch;
- @Override
- public String toString() {
- return name().toLowerCase(Locale.ROOT);
+ public Content() {};
+
+ public Content(int version, Long brokerEpoch) {
+ this.version = version;
+ this.brokerEpoch = brokerEpoch;
}
}
@@ -73,10 +76,8 @@ public class CleanShutdownFileHandler {
FileOutputStream os = new FileOutputStream(cleanShutdownFile);
BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(os, StandardCharsets.UTF_8));
try {
- Map<String, String> payload = new HashMap<>();
- payload.put(Fields.VERSION.toString(), Integer.toString(version));
- payload.put(Fields.BROKER_EPOCH.toString(), Long.toString(brokerEpoch));
- bw.write(new ObjectMapper().writeValueAsString(payload));
+ Content content = new Content(version, brokerEpoch);
+ bw.write(Json.encodeAsString(content));
bw.flush();
os.getFD().sync();
} finally {
@@ -85,17 +86,16 @@ public class CleanShutdownFileHandler {
}
}
- public long read() {
- long brokerEpoch = -1L;
+ @SuppressWarnings("unchecked")
+ public OptionalLong read() {
try {
String text = Utils.readFileAsString(cleanShutdownFile.toPath().toString());
- Map<String, String> content = new ObjectMapper().readValue(text, HashMap.class);
-
- brokerEpoch = Long.parseLong(content.getOrDefault(Fields.BROKER_EPOCH.toString(), "-1L"));
+ Content content = Json.parseStringAs(text, Content.class);
+ return OptionalLong.of(content.brokerEpoch);
} catch (Exception e) {
- logger.warn("Fail to read the clean shutdown file in " + cleanShutdownFile.toPath() + ":" + e);
+ logger.debug("Fail to read the clean shutdown file in " + cleanShutdownFile.toPath() + ":" + e);
+ return OptionalLong.empty();
}
- return brokerEpoch;
}
public void delete() throws Exception {
diff --git a/core/src/test/java/kafka/log/CleanShutdownFileHandlerTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/checkpoint/CleanShutdownFileHandlerTest.java
similarity index 63%
rename from core/src/test/java/kafka/log/CleanShutdownFileHandlerTest.java
rename to storage/src/test/java/org/apache/kafka/storage/internals/checkpoint/CleanShutdownFileHandlerTest.java
index 977bcb758e3..e2ba45db681 100644
--- a/core/src/test/java/kafka/log/CleanShutdownFileHandlerTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/checkpoint/CleanShutdownFileHandlerTest.java
@@ -15,13 +15,19 @@
* limitations under the License.
*/
-package kafka.log;
+package org.apache.kafka.storage.internals.checkpoint;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import java.io.BufferedWriter;
import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
+import java.util.OptionalLong;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -38,7 +44,7 @@ class CleanShutdownFileHandlerTest {
CleanShutdownFileHandler cleanShutdownFileHandler = new CleanShutdownFileHandler(logDir.getPath());
assertDoesNotThrow(() -> cleanShutdownFileHandler.write(10L));
assertTrue(cleanShutdownFileHandler.exists());
- assertEquals(10L, cleanShutdownFileHandler.read());
+ assertEquals(OptionalLong.of(10L), cleanShutdownFileHandler.read());
assertDoesNotThrow(() -> cleanShutdownFileHandler.delete());
assertFalse(cleanShutdownFileHandler.exists());
}
@@ -52,6 +58,24 @@ class CleanShutdownFileHandlerTest {
assertTrue(cleanShutdownFileHandler.exists());
assertDoesNotThrow(() -> cleanShutdownFileHandler.delete());
assertFalse(cleanShutdownFileHandler.exists());
- assertEquals(-1L, cleanShutdownFileHandler.read());
+ assertEquals(OptionalLong.empty(), cleanShutdownFileHandler.read());
+ }
+
+ @Test
+ public void testCleanShutdownFileCanParseWithUnknownFields() throws IOException {
+ File logDir;
+ logDir = assertDoesNotThrow(() -> Files.createTempDirectory("kafka-cleanShutdownFile").toFile());
+ CleanShutdownFileHandler cleanShutdownFileHandler = new CleanShutdownFileHandler(logDir.getPath());
+
+ FileOutputStream os = new FileOutputStream(cleanShutdownFileHandler.cleanShutdownFile);
+ BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(os, StandardCharsets.UTF_8));
+ bw.write("{\"version\":0,\"brokerEpoch\":10,\"unknown\":123}");
+ bw.flush();
+ os.getFD().sync();
+
+ assertTrue(cleanShutdownFileHandler.exists());
+ assertEquals(OptionalLong.of(10L), cleanShutdownFileHandler.read());
+ assertDoesNotThrow(() -> cleanShutdownFileHandler.delete());
+ assertFalse(cleanShutdownFileHandler.exists());
}
}
\ No newline at end of file