Merge pull request #2150 from chenharryhua/0.13

0.13
chenharryhua · Apr 24, 2022 · 2fd8959 · 2fd8959
2 parents f2bf885 + aa62baf
commit 2fd8959
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 13 deletions.
diff --git a/kafka/src/test/scala/mtest/kafka/stream/KafkaStreamingTest.scala b/kafka/src/test/scala/mtest/kafka/stream/KafkaStreamingTest.scala
@@ -34,14 +34,14 @@ object KafkaStreamingData {
   val s1Data: Stream[IO, ProducerRecords[Int, StreamOne]] = Stream
     .emits(
       List(
-        ProducerRecord[Int, StreamOne](s1Topic.topicName.value, 101, StreamOne("na", -1)),
-        ProducerRecord[Int, StreamOne](s1Topic.topicName.value, 102, StreamOne("na", -1)),
-        ProducerRecord[Int, StreamOne](s1Topic.topicName.value, 103, StreamOne("na", -1)),
-        ProducerRecord[Int, StreamOne](s1Topic.topicName.value, 1, StreamOne("a", 0)),
-        ProducerRecord[Int, StreamOne](s1Topic.topicName.value, 2, StreamOne("b", 1)),
-        ProducerRecord[Int, StreamOne](s1Topic.topicName.value, 3, StreamOne("c", 2)),
-        ProducerRecord[Int, StreamOne](s1Topic.topicName.value, 201, StreamOne("d", 3)),
-        ProducerRecord[Int, StreamOne](s1Topic.topicName.value, 202, StreamOne("e", 4))
+        s1Topic.fs2ProducerRecord(101, StreamOne("na", -1)),
+        s1Topic.fs2ProducerRecord(102, StreamOne("na", -1)),
+        s1Topic.fs2ProducerRecord(103, StreamOne("na", -1)),
+        s1Topic.fs2ProducerRecord(1, StreamOne("a", 0)),
+        s1Topic.fs2ProducerRecord(2, StreamOne("b", 1)),
+        s1Topic.fs2ProducerRecord(3, StreamOne("c", 2)),
+        s1Topic.fs2ProducerRecord(201, StreamOne("d", 3)),
+        s1Topic.fs2ProducerRecord(202, StreamOne("e", 4))
       ).map(ProducerRecords.one))
     .covary[IO]
 

diff --git a/pipes/src/main/scala/com/github/chenharryhua/nanjin/terminals/NJPath.scala b/pipes/src/main/scala/com/github/chenharryhua/nanjin/terminals/NJPath.scala
@@ -7,7 +7,7 @@ import eu.timepit.refined.api.Refined
 import io.circe.generic.JsonCodec
 import io.circe.refined.*
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{LocatedFileStatus, Path}
 import org.apache.parquet.hadoop.util.{HadoopInputFile, HadoopOutputFile}
 
 import java.net.URI
@@ -51,10 +51,11 @@ final case class NJPath private (root: PathRoot, segments: List[PathSegment]) {
 }
 object NJPath {
 
-  def apply(root: PathRoot): NJPath = NJPath(root, Nil)
-  def apply(hp: Path): NJPath       = NJPath(PathRoot.unsafeFrom(hp.toString))
-  def apply(uri: URI): NJPath       = NJPath(PathRoot.unsafeFrom(uri.toASCIIString))
-  def apply(s3: S3Path): NJPath     = NJPath(PathRoot.unsafeFrom(s3.s3a))
+  def apply(root: PathRoot): NJPath         = NJPath(root, Nil)
+  def apply(hp: Path): NJPath               = apply(PathRoot.unsafeFrom(hp.toString))
+  def apply(uri: URI): NJPath               = apply(PathRoot.unsafeFrom(uri.toASCIIString))
+  def apply(s3: S3Path): NJPath             = apply(PathRoot.unsafeFrom(s3.s3a))
+  def apply(lfs: LocatedFileStatus): NJPath = apply(lfs.getPath)
 
   implicit val showNJPath: Show[NJPath] = _.pathStr