use UUIDv7 for Iceberg filenames

stream-loader-tests/src/main/scala/com/adform/streamloader/util/UuidExtensions.scala → stream-loader-core/src/main/scala/com/adform/streamloader/util/UuidExtensions.scala

@@ -9,6 +9,7 @@
 package com.adform.streamloader.util
 
 import java.nio.ByteBuffer
+import java.security.SecureRandom
 import java.util.UUID
 
 object UuidExtensions {
@@ -20,4 +21,24 @@ object UuidExtensions {
       bb.array
     }
   }
+
+  private val random = new SecureRandom
+
+  def randomUUIDv7(): UUID = {
+    val value = new Array[Byte](16)
+    random.nextBytes(value)
+
+    val timestamp = ByteBuffer.allocate(8)
+    timestamp.putLong(System.currentTimeMillis)
+    System.arraycopy(timestamp.array, 2, value, 0, 6)
+
+    value(6) = ((value(6) & 0x0f) | 0x70).toByte
+    value(8) = ((value(8) & 0x3f) | 0x80).toByte
+
+    val buf = ByteBuffer.wrap(value)
+    val high = buf.getLong
+    val low = buf.getLong
+
+    new UUID(high, low)
+  }
 }

stream-loader-core/src/test/scala/com/adform/streamloader/util/UuidExtensionsTest.scala

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2020 Adform
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+package com.adform.streamloader.util
+
+import com.adform.streamloader.util.UuidExtensions.randomUUIDv7
+import org.scalatest.funspec.AnyFunSpec
+import org.scalatest.matchers.should.Matchers
+
+class UuidExtensionsTest extends AnyFunSpec with Matchers {
+
+  it("should generate unique UUIDv7 values") {
+    val uuids = (0 until 100).map(_ => randomUUIDv7())
+    uuids should contain theSameElementsInOrderAs uuids.distinct
+  }
+
+  it("should generate alphabetically sorted UUIDv7 values") {
+    val uuids = (0 until 10)
+      .map(_ => {
+        randomUUIDv7()
+        Thread.sleep(5) // the v7 spec guarantees ordering in a 1ms scope
+      })
+      .map(_.toString)
+
+    uuids should contain theSameElementsInOrderAs uuids.sorted
+  }
+}

stream-loader-iceberg/src/main/scala/com/adform/streamloader/iceberg/IcebergRecordBatcher.scala

@@ -12,12 +12,12 @@ import com.adform.streamloader.model.{StreamRange, StreamRecord}
 import com.adform.streamloader.sink.batch.{RecordBatch, RecordBatchBuilder, RecordBatcher, RecordFormatter}
 import com.adform.streamloader.sink.file.{FileStats, MultiFileCommitStrategy}
 import com.adform.streamloader.util.TimeProvider
+import com.adform.streamloader.util.UuidExtensions.randomUUIDv7
 import org.apache.iceberg.data.{GenericAppenderFactory, Record => IcebergRecord}
 import org.apache.iceberg.io.DataWriteResult
 import org.apache.iceberg.{FileFormat, PartitionKey, Table}
 
 import java.time.Duration
-import java.util.UUID
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
 
@@ -41,7 +41,7 @@ class IcebergRecordBatchBuilder(
     private val startTimeMillis = timeProvider.currentMillis
     private var recordCount = 0L
     private val dataWriter = {
-      val filename = fileFormat.addExtension(UUID.randomUUID().toString)
+      val filename = fileFormat.addExtension(randomUUIDv7().toString)
       val path = table.locationProvider().newDataLocation(table.spec(), pk, filename)
       val output = table.io().newOutputFile(path)