You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Steven Schlansker (JIRA)" <ji...@apache.org> on 2018/04/09 19:24:00 UTC
[jira] [Created] (KAFKA-6767) OffsetCheckpoint write assumes parent
directory exists
Steven Schlansker created KAFKA-6767:
----------------------------------------
Summary: OffsetCheckpoint write assumes parent directory exists
Key: KAFKA-6767
URL: https://issues.apache.org/jira/browse/KAFKA-6767
Project: Kafka
Issue Type: Bug
Components: streams
Affects Versions: 1.1.0
Reporter: Steven Schlansker
We run Kafka Streams with RocksDB state stores on ephemeral disks (i.e. if an instance dies it is created from scratch, rather than reusing the existing RocksDB.)
We routinely see:
{code:java}
2018-04-09T19:14:35.004Z WARN <> [chat-0319e3c3-d8b2-4c60-bd69-a8484d8d4435-StreamThread-1] o.a.k.s.p.i.ProcessorStateManager - task [0_11] Failed to write offset checkpoint file to /mnt/mesos/sandbox/storage/chat/0_11/.checkpoint: {}
java.io.FileNotFoundException: /mnt/mesos/sandbox/storage/chat/0_11/.checkpoint.tmp (No such file or directory)
at java.io.FileOutputStream.open0(Native Method)
at java.io.FileOutputStream.open(FileOutputStream.java:270)
at java.io.FileOutputStream.<init>(FileOutputStream.java:213)
at java.io.FileOutputStream.<init>(FileOutputStream.java:162)
at org.apache.kafka.streams.state.internals.OffsetCheckpoint.write(OffsetCheckpoint.java:78)
at org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint(ProcessorStateManager.java:320)
at org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:314)
at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:307)
at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:297)
at org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:67)
at org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:357)
at org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:347)
at org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:403)
at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:994)
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:811)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:750)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:720){code}
Inspecting the state store directory, I can indeed see that {{chat/0_11}} does not exist (although many other partitions do).
Looking at the OffsetCheckpoint write method, it seems to try to open a new checkpoint file without first ensuring that the parent directory exists.
{code:java}
public void write(final Map<TopicPartition, Long> offsets) throws IOException {
// if there is no offsets, skip writing the file to save disk IOs
if (offsets.isEmpty()) {
return;
}
synchronized (lock) {
// write to temp file and then swap with the existing file
final File temp = new File(file.getAbsolutePath() + ".tmp");{code}
Either the OffsetCheckpoint class should initialize the directories if needed, or some precondition of it being called should ensure that is the case.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)