# Failing Kafka when KafkaStream has set own TimestampExtractor

**URL:** <https://forum.confluent.io/t/failing-kafka-when-kafkastream-has-set-own-timestampextractor/1755>\
**Category:** Kafka Streams\
**Created:** [12 May 2021 10:49 UTC](https://forum.confluent.io/t/failing-kafka-when-kafkastream-has-set-own-timestampextractor/1755 "2021-05-12T10:49:42Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![programista4k](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/programista4k/32/767_2.png) [@programista4k](https://forum.confluent.io/u/programista4k)\
**Post date:** [12 May 2021 10:49 UTC](https://forum.confluent.io/t/failing-kafka-when-kafkastream-has-set-own-timestampextractor/1755/1 "2021-05-12T10:49:42Z")

</div>

Hello frens,  
I get such exception from Kafka while runtime:

> Failed to clean up log for statefull\_app\_id-raw-trades-changelog-0 in dir C:\Kafka\kafka\_2.13-2.7.0\tmp\kafka-logs due to IOException (kafka.server.LogDirFailureChannel)  
> java.nio.file.FileSystemException: C:\Kafka\kafka\_2.13-2.7.0\tmp\kafka-logs\statefull\_app\_id-raw-trades-changelog-0\00000000000000000000.timeindex.cleaned → C:\Kafka\kafka\_2.13-2.7.0\tmp\kafka-logs\statefull\_app\_id-raw-trades-changelog-0\00000000000000000000.timeindex.swap: The process cannot access the file because it is in use by another process  
> at java.base/sun.nio.fs.WindowsException.translateToIOException(WindowsException.java:92)  
> at java.base/sun.nio.fs.WindowsException.rethrowAsIOException(WindowsException.java:103)  
> at java.base/sun.nio.fs.WindowsFileCopy.move(WindowsFileCopy.java:395)  
> at java.base/sun.nio.fs.WindowsFileSystemProvider.move(WindowsFileSystemProvider.java:292)  
> at java.base/java.nio.file.Files.move(Files.java:1426)  
> at org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:917)  
> at kafka.log.AbstractIndex.renameTo(AbstractIndex.scala:211)  
> at kafka.log.LazyIndex$IndexValue.renameTo(LazyIndex.scala:155)  
> at kafka.log.LazyIndex.$anonfun$renameTo$1(LazyIndex.scala:79)  
> at kafka.log.LazyIndex.renameTo(LazyIndex.scala:79)  
> at kafka.log.LogSegment.changeFileSuffixes(LogSegment.scala:497)  
> at kafka.log.Log.$anonfun$replaceSegments$4(Log.scala:2338)  
> at kafka.log.Log.$anonfun$replaceSegments$4$adapted(Log.scala:2338)  
> at scala.collection.immutable.List.foreach(List.scala:333)  
> at kafka.log.Log.replaceSegments(Log.scala:2338)  
> at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:614)  
> at kafka.log.Cleaner.$anonfun$doClean$6(LogCleaner.scala:539)  
> at kafka.log.Cleaner.doClean(LogCleaner.scala:538)  
> at kafka.log.Cleaner.clean(LogCleaner.scala:512)  
> at kafka.log.LogCleaner$CleanerThread.cleanLog(LogCleaner.scala:381)  
> at kafka.log.LogCleaner$CleanerThread.cleanFilthiestLog(LogCleaner.scala:353)  
> at kafka.log.LogCleaner$CleanerThread.tryCleanFilthiestLog(LogCleaner.scala:333)  
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:322)  
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:96)  
> Suppressed: java.nio.file.FileSystemException: C:\Kafka\kafka\_2.13-2.7.0\tmp\kafka-logs\statefull\_app\_id-raw-trades-changelog-0\00000000000000000000.timeindex.cleaned → C:\Kafka\kafka\_2.13-2.7.0\tmp\kafka-logs\statefull\_app\_id-raw-trades-changelog-0\00000000000000000000.timeindex.swap: **The process cannot access the file because it is in use by another process**  
> at java.base/sun.nio.fs.WindowsException.translateToIOException(WindowsException.java:92)  
> at java.base/sun.nio.fs.WindowsException.rethrowAsIOException(WindowsException.java:103)  
> at java.base/sun.nio.fs.WindowsFileCopy.move(WindowsFileCopy.java:309)  
> at java.base/sun.nio.fs.WindowsFileSystemProvider.move(WindowsFileSystemProvider.java:292)  
> at java.base/java.nio.file.Files.move(Files.java:1426)  
> at org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:914)  
> … 18 more

this is my stream:

```auto
KStream<String, CryptoTradeRecord> tradeStream = streamsBuilder
        .addStateStore(storeBuilder)
        .stream(KafkaTopics.CRYPTO_TRADES,
                Consumed.with(stringSerde, cryptoTradeSerde)
                        .withTimestampExtractor(myTimestampExtractor));

KStream<String, CryptoTradeRecord> distinctTradeStream = tradeStream
        .filter((k, v) -> cryptoTradeRecordValidator.validate(v).isValid)
        .transformValues(() -> new CryptoTradeRecordDistinctTransformer(storeName), storeName)
        .filter((k, v) -> v != null);

```

and this is the TimestampExtractor:

```auto
@Override
public long extract(ConsumerRecord<Object, Object> consumerRecord, long l) {
    CryptoTradeRecord cryptoTradeRecord = (CryptoTradeRecord) consumerRecord.value();
    return cryptoTradeRecord .getEventTimestamp();
}

```

It fails after 1000 to 2000 messages.

---

<div class="post-metadata">

**Author:** ![Bruno](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/bruno/32/138_2.png) [@Bruno](https://forum.confluent.io/u/Bruno)\
**Post date:** [12 May 2021 11:35 UTC](https://forum.confluent.io/t/failing-kafka-when-kafkastream-has-set-own-timestampextractor/1755/2 "2021-05-12T11:35:56Z")

</div>

This does not look like a Kafka Streams issue. The log cleaner on the brokers throws the exception. The brokers are agnostic to who is reading the topics. Try to post your question in the Ops category.  
Note that Windows is not very well supported for Kafka.

---

<div class="post-metadata">

**Author:** ![programista4k](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/programista4k/32/767_2.png) [@programista4k](https://forum.confluent.io/u/programista4k)\
**Post date:** [12 May 2021 11:44 UTC](https://forum.confluent.io/t/failing-kafka-when-kafkastream-has-set-own-timestampextractor/1755/3 "2021-05-12T11:44:43Z")

</div>

I have fixed it by shrinking my timestamp by dividing it with 1000.  
Before I used nano timestamp like:  
`1620818525470329111` now I use  
`1620818525470329` and now it works.  
But still doesn’t understand why the nano timestamp was making failures.