What's wrong with flow join the way i use

JackyChan · March 20, 2021, 3:26pm

hello there, i am a newbie to Akka. I learn Akka Stream recently, and i see a method join within flow. I try to run an example with it. The code like following:

val f1 = Flow[Int].fold(0)(_+_)
val f2 = Flow[Int].fold(0)(_+_)
f1.join(f2)

Run it and then produce NullPointerException as following:

java.lang.NullPointerException
  akka.stream.Attributes$SourceLocation.locationName$lzycompute(Attributes.scala:319)
  akka.stream.Attributes$SourceLocation.locationName(Attributes.scala:311)
  akka.stream.Attributes$SourceLocation.toString(Attributes.scala:322)
  pprint.Walker.$anonfun$treeify$9(Walker.scala:108)
  pprint.Renderer.str$lzycompute$1(Renderer.scala:132)
  pprint.Renderer.str$1(Renderer.scala:130)
  pprint.Renderer.rec(Renderer.scala:134)
  pprint.Renderer.rec(Renderer.scala:41)
  pprint.Renderer.rec(Renderer.scala:41)
  pprint.Renderer.rec(Renderer.scala:41)
  pprint.Renderer.rec(Renderer.scala:41)
  pprint.Renderer.rec(Renderer.scala:41)
  pprint.Renderer.rec(Renderer.scala:41)
  pprint.PPrinter.tokenize(PPrinter.scala:105)
  ammonite.repl.FullReplAPI$Internal.print(FullReplAPI.scala:106)
  ammonite.repl.FullReplAPI$Internal.print$(FullReplAPI.scala:61)
  ammonite.repl.FullReplAPI$$anon$1.print(FullReplAPI.scala:34)

I search for help but no lucky. Any detail about this? please.

JackyChan · March 20, 2021, 5:33pm

It seems to be a toString problem. try to run f1.join(f2).toString would produce the exception. Like following:

f1.toString
res62: String = "Flow(FlowShape(Fold.in(324764712),Fold.out(1261141566)))"

f2.toString
res63: String = "Flow(FlowShape(Fold.in(1963435950),Fold.out(2060916294)))"

f1.join(f2).toString
java.lang.NullPointerException
  akka.stream.Attributes$SourceLocation.locationName$lzycompute(Attributes.scala:319)
  akka.stream.Attributes$SourceLocation.locationName(Attributes.scala:311)
  akka.stream.Attributes$SourceLocation.toString(Attributes.scala:322)
  java.lang.String.valueOf(String.java:2994)
  java.lang.StringBuilder.append(StringBuilder.java:131)
  scala.collection.IterableOnceOps.addString(IterableOnce.scala:1191)
  scala.collection.IterableOnceOps.addString$(IterableOnce.scala:1186)
  scala.collection.AbstractIterable.addString(Iterable.scala:920)
  scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1136)
  scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1134)
  scala.collection.AbstractIterable.mkString(Iterable.scala:920)
  scala.collection.Iterable.toString(Iterable.scala:76)
  scala.collection.Iterable.toString$(Iterable.scala:76)
  scala.collection.Seq.toString(Seq.scala:49)
  scala.collection.Seq.toString$(Seq.scala:49)
  scala.collection.AbstractSeq.toString(Seq.scala:1154)
  java.lang.String.valueOf(String.java:2994)
  java.lang.StringBuilder.append(StringBuilder.java:131)
  scala.collection.IterableOnceOps.addString(IterableOnce.scala:1191)
  scala.collection.IterableOnceOps.addString$(IterableOnce.scala:1186)
  scala.collection.AbstractIterator.addString(Iterator.scala:1279)
  scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1136)
  scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1134)
  scala.collection.AbstractIterator.mkString(Iterator.scala:1279)
  scala.runtime.ScalaRunTime$._toString(ScalaRunTime.scala:159)
  akka.stream.Attributes.toString(Attributes.scala:36)
  java.lang.String.valueOf(String.java:2994)
  java.lang.StringBuilder.append(StringBuilder.java:131)
  scala.collection.IterableOnceOps.addString(IterableOnce.scala:1191)
  scala.collection.IterableOnceOps.addString$(IterableOnce.scala:1186)
  scala.collection.AbstractIterator.addString(Iterator.scala:1279)
  scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1136)
  scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1134)
  scala.collection.AbstractIterator.mkString(Iterator.scala:1279)
  scala.runtime.ScalaRunTime$._toString(ScalaRunTime.scala:159)
  akka.stream.impl.PushAttributes.toString(TraversalBuilder.scala:178)
  java.lang.String.valueOf(String.java:2994)
  java.lang.StringBuilder.append(StringBuilder.java:131)
  scala.collection.IterableOnceOps.addString(IterableOnce.scala:1191)
  scala.collection.IterableOnceOps.addString$(IterableOnce.scala:1186)
  scala.collection.AbstractIterator.addString(Iterator.scala:1279)
  scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1136)
  scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1134)
  scala.collection.AbstractIterator.mkString(Iterator.scala:1279)
  scala.runtime.ScalaRunTime$._toString(ScalaRunTime.scala:159)
  akka.stream.impl.Concat.toString(TraversalBuilder.scala:93)
  java.lang.String.valueOf(String.java:2994)
  java.lang.StringBuilder.append(StringBuilder.java:131)
  scala.collection.IterableOnceOps.addString(IterableOnce.scala:1191)
  scala.collection.IterableOnceOps.addString$(IterableOnce.scala:1186)
  scala.collection.AbstractIterator.addString(Iterator.scala:1279)
  scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1136)
  scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1134)
  scala.collection.AbstractIterator.mkString(Iterator.scala:1279)
  scala.runtime.ScalaRunTime$._toString(ScalaRunTime.scala:159)
  akka.stream.impl.Concat.toString(TraversalBuilder.scala:93)
  java.lang.String.valueOf(String.java:2994)
  java.lang.StringBuilder.append(StringBuilder.java:131)
  scala.collection.IterableOnceOps.addString(IterableOnce.scala:1194)
  scala.collection.IterableOnceOps.addString$(IterableOnce.scala:1186)
  scala.collection.AbstractIterator.addString(Iterator.scala:1279)
  scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1136)
  scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1134)
  scala.collection.AbstractIterator.mkString(Iterator.scala:1279)
  scala.runtime.ScalaRunTime$._toString(ScalaRunTime.scala:159)
  akka.stream.impl.Concat.toString(TraversalBuilder.scala:93)
  java.lang.String.valueOf(String.java:2994)
  java.lang.StringBuilder.append(StringBuilder.java:131)
  scala.collection.IterableOnceOps.addString(IterableOnce.scala:1194)
  scala.collection.IterableOnceOps.addString$(IterableOnce.scala:1186)
  scala.collection.AbstractIterator.addString(Iterator.scala:1279)
  scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1136)
  scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1134)
  scala.collection.AbstractIterator.mkString(Iterator.scala:1279)
  scala.runtime.ScalaRunTime$._toString(ScalaRunTime.scala:159)
  akka.stream.impl.LinearTraversalBuilder.toString(TraversalBuilder.scala:735)
  java.lang.String.valueOf(String.java:2994)
  java.lang.StringBuilder.append(StringBuilder.java:131)
  scala.collection.IterableOnceOps.addString(IterableOnce.scala:1191)
  scala.collection.IterableOnceOps.addString$(IterableOnce.scala:1186)
  scala.collection.AbstractIterator.addString(Iterator.scala:1279)
  scala.collection.IterableOnceOps.mkString(IterableOnce.scala:1136)
  scala.collection.IterableOnceOps.mkString$(IterableOnce.scala:1134)
  scala.collection.AbstractIterator.mkString(Iterator.scala:1279)
  scala.runtime.ScalaRunTime$._toString(ScalaRunTime.scala:159)
  akka.stream.scaladsl.RunnableGraph.toString(Flow.scala:727)
  ammonite.$sess.cmd64$.<clinit>(cmd64.sc:1)

seglo · March 22, 2021, 2:06pm

I wasn’t able to reproduce this with Akka 2.6.13. What version are you using?

octonato · March 22, 2021, 6:59pm

For the record, also reported on GitHub

Thanks @JackyChan

There is a PR under way. Extra catch-ignore for source attributes in case not possible by johanandren · Pull Request #30139 · akka/akka · GitHub

Topic		Replies	Views
Use akka-stream-typed from play? Akka Streams & Alpakka akka-typed	0	742	December 9, 2021
Using Flow as Sink/Source Akka akka-http , streams	2	908	October 16, 2020
Understanding Akka Streams - play-socket.io Akka	2	900	January 7, 2020
Java examples and documentations Akka Streams & Alpakka	6	650	September 25, 2021
AkkaStream. Proper flow decomposition Akka Streams & Alpakka	4	889	September 18, 2018

What's wrong with flow join the way i use

Related topics