Questions tagged [akka-stream]
Akka implementation for handling streaming data on the JVM
1,692
questions
0
votes
1
answer
42
views
Akka stream stops processing data
When I run the below stream it does not receive any subsequent data once the stream runs.
final long HOUR = 3600000;
final long PAST_HOUR = System.currentTimeMillis()-HOUR;
private final ...
0
votes
1
answer
30
views
Using ElasticsearchSource to filter results in Elastic
Here I'm using this code to read data from an index where timestamp is >= 1656055230028:
Source<ReadResult<Map<String, Object>>, NotUsed> source =
...
0
votes
1
answer
80
views
Akka-stream implicitly change execution thread
During development of app with java+akka-stream I faced one strange behaviour.
According my understanding of akka-stream's documentation processing thread should not change without explicit ...
2
votes
1
answer
84
views
How to use an Actor based source with an Akka Graph?
I'm attempting to send data via an actor to a runnable graph that contains a fan out.
I define the source as :
final Source<Integer, ActorRef> integerSource =
Source.actorRef(
...
1
vote
0
answers
24
views
Response entity was not subscribed after 100 seconds. Make sure to read the response entity body or call `discardBytes()` on it
I am building a scala application. Within the application, we are making a call to an external service, and fetching the data.
When I am hitting the endpoint of this external service using the
...
1
vote
1
answer
44
views
Use akka to stream contents of streaming file as a HttpResponse
I am reading parts of large file via a Java FileInputStream and would like to stream it's content back to the client (in the form of an akka HttpResponse). I am wondering if this is possible, and how ...
0
votes
0
answers
12
views
Is it possible to set various number of threads during a run ? (change the number of Akka threads dynamically)
In order to control the number of threads of Akka,
I would like to know if there is a possibility to change the number of Akka's threads dynamically-
When the program is already running,
is it ...
0
votes
0
answers
14
views
[Akka-stream][Errorr]: java.lang.IllegalArgumentException: [Partition.out0] is already connected
def partitionDocumentStorage(successMessage: SuccessMessage) =
successMessage.message.request match {
case InlineDocumentRequest(_) => 0
case ImportDocumentRequest(_) => 1
...
0
votes
0
answers
71
views
akka kafka consumer poll speed backpressure
The project uses akka stream kafka, and one process is commencing several kafka topics. When I receive a message from kafka, I am processing the business logic of pre-processing and updating to ...
2
votes
1
answer
107
views
java.lang.NoSuchMethodError: 'void akka.actor.ExtensionId.$init$(akka.actor.ExtensionId)'
I am simply trying to implement a little sample REST endpoint in a scala project using akka.
The code looks as follows
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{ContentTypes, ...
0
votes
0
answers
19
views
Why does adding an explicit type parameter to `Source.actorRef` cause "missing parameter type for expanded function" error?
This code, adapted from a Source.actorRef documentation, works fine:
import akka.actor.ActorSystem
import akka._
import akka.actor.ActorRef
import akka.stream.scaladsl._
implicit val system = ...
0
votes
1
answer
41
views
Akka streams - why is an ActorSystem accepted as a Materalizer only when it is marked as implicit?
I'm looking at the akka streams quickstart tutorial and I wanted to understand how a piece of code works. The following code from the example prints out the values from 1 to 100 to the console:
import ...
0
votes
1
answer
37
views
Dynamic Akka streams sink
I want to build a graph in Akka streams where the source is a Kafka topic (topic_a) and the sink is topic_b (always) and, depending on the message's data, also topic_c (the output message will be ...
0
votes
1
answer
49
views
Akka Source that emits when another Sink receives
I have a source a that emits values into a sink b.
Now I want to have another source c that emits a value, everytime b receives an event.
My idea was to use another sink d that can be used as a ...
0
votes
1
answer
48
views
When not to use NotUsed in Akka Stream
I have seen a lot of Akka stream examples that use NotUsed, but I have not been able to find one where something other than NotUsed was used. I tried creating a source that does not have NotUsed, but ...
1
vote
1
answer
48
views
Kafka to MongoDB with Akka Streams
I'm pretty new to the Akka streams technology and I didn't find any source of information regarding my use case.
I want to use Akka streams to consume messages from a Kafka topic, process them (while ...
0
votes
0
answers
86
views
Backpressuring on async Flow in Akka Streams
I have this flow component in Akka streams
object ExecutionFlow {
def apply(
execute: In => Future[Out],
parallelism: Int)(implicit ec: ExecutionContext): Flow[In, Out, ...
1
vote
2
answers
140
views
Akka stream hangs when starting more than 15 external processes using ProcessBuilder
I'm building an app that has the following flow:
There is a source of items to process
Each item should be processed by external command (it'll be ffmpeg in the end but for this simple reproducible ...
0
votes
0
answers
56
views
Writing POST API for Multi Part file Upload using Scala Akka framework
I am implementing POST api for multipart file upload using scala akka framework ,with following method definitition
def postFileUploadResponse(respond: Resource.PostFileUploadResponse.type)(filetype: ...
0
votes
1
answer
70
views
Akka, chaining Futures
I have three methods which return future, how do I properly chain them so that output from first one is passed to the second and second's return to third and finally send them all of the above data to ...
0
votes
0
answers
39
views
Akka spawn thread using Semahpore
I'm new to multithreading, I'm using Akka Stream to process few files concurrently, based on the below code, multiple threads
are spawned, the memory gets full and the OS kills the process.
I also ...
1
vote
0
answers
33
views
RestartFlow.onFailuresWithBackoff not restarting on failure
We are encountering an unexpected behavior from:
RestartFlow.onFailuresWithBackoff(restartSetting)(flowFactory: () => Flow[In, Out, _])
See RestartFlow.onFailuresWithBackoff.
We're using:
akka:2....
0
votes
1
answer
80
views
How to split a stream into two flows using GraphDSL
I'm learning akka streams and as an example I'm attempting to split data into two separate flows depending on the structure of JSON received from a source using the GraphDSL.
Here is the complete code ...
1
vote
0
answers
50
views
Trying to understand this Graph-DSL example (a simple FlowShape)
Given:
val merge = builder.add(Merge[Command](2))
msgToDomainCommand ~> merge.in(0)
domainCommand ~> merge.in(1)
merge ~> routeToSession
socket ~> eventToWebSocketsShape
// expose ...
1
vote
1
answer
87
views
Akka stream best practice for dynamic Source and Flow controlled by websocket messages
I'm trying to understand what is the best way to implement with akka stream and alpakka the following scenario (simplified):
The frontend opens a websocket connection with the backend
Backend should ...
1
vote
2
answers
154
views
How to handle java.util.NoSuchElementException: reduce over empty stream akka stream?
I have a filter to filter out incorrect items in a stream. In some edge cases this could result in all items being filtered out. When this happens the stream fails with error -
java.util....
1
vote
1
answer
281
views
Why does auto-commit enabled Kafka client commit latest produced message's offset during consumer close even if the message was not consumed yet?
TLDR:
Is committing produced message's offset as consumed (even if it wasn't) expected behavior for auto-commit enabled Kafka clients? (for the applications that consuming and producing the same ...
0
votes
1
answer
71
views
Pattern for separating akka streams Flows
I'm taking an Akka Streams Udemy course and this code is provided :
package multiflow;
import akka.NotUsed;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.javadsl.Behaviors;
import akka....
1
vote
0
answers
83
views
alpakka s3 source cause akka stream to crash with TcpIdleTimeoutException: TCP idle-timeout encountered on connection
I am reading a file from S3 using alpakka s3 client
import akka.stream.alpakka.s3.scaladsl.S3
S3.download(bckt, bcktKey).flatMapConcat{
case Some((file, _)) =>
Foo
}.mapAsync(1){ f=>
...
0
votes
1
answer
83
views
How to retrieve messages from Alpakka Mqtt Streaming client?
I was following document for writing a Mqtt client subscriber using alpakka.
https://doc.akka.io/docs/alpakka/3.0.4/mqtt-streaming.html?_ga=2.247958340.274298740.1642514263-524322027.1627936487
After ...
1
vote
1
answer
111
views
Akka streams RestartSink doesn't seem to be restarting during failures
I am playing around with handling errors in akka streams with restartable sources & sinks.
object Main extends App {
implicit val system: ActorSystem = ActorSystem("akka-streams-system"...
0
votes
1
answer
102
views
How can an event sourced entity to subscribe to state changes in another entity?
I have an events-sourced entity (C) that needs to change its state in response to state changes in another entity of a different type (P). The logic to whether the state of C should actually change is ...
0
votes
0
answers
73
views
object stream is not a member of package akka
I've below build.sbt file -
ThisBuild / version := "0.1.0-SNAPSHOT"
ThisBuild / scalaVersion := "2.13.7"
val akkaVersion = "2.6.18"
lazy val root = (project in file(&...
0
votes
1
answer
64
views
Akka Streams split Stream for Error handling
I'm using akka http and streams to fulfill API requests.
When the request is invalid I want to return a 400 and if it's valid, i want to proceed with the computation and return the result afterwards.
...
0
votes
0
answers
29
views
How to remove deprecated AkkaSSLConfig using createSSLEgine
I'm trying to remove AkkaSSLConfig from code usage. Below is my code.
import com.typesafe.sslconfig.akka.AkkaSSLConfig
import akka.http.scaladsl.HttpsConnectionContext
val sslConfig = ...
-1
votes
0
answers
56
views
Akka Streams, StreamConverters and System.in [duplicate]
I'm using Scala 2.13.7 with Akka Streams 2.6.17.
I have this very basic Akka Stream. It takes what the user types (System.in), applies toUpperCase function and finally the string is printed using ...
0
votes
2
answers
73
views
Akka streams websocket stream things to a Sink.seq ends with exception SubscriptionWithCancelException$StageWasCompleted
I'm failing to materialize the Sink.seq, when it comes time to materialize I fail with this exception
akka.stream.SubscriptionWithCancelException$StageWasCompleted$:
Here is the full source code on ...
1
vote
0
answers
57
views
Handling downstream IOExceptions in Akka HTTP server
How do I catch downstream exceptions such as java.io.IOException when returning responses?
I tried using CancellationStrategy.FailStage attribute but stage still completes successfully when using ...
1
vote
1
answer
126
views
How to consume HttpResponse in Akka Http
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpMethods, HttpRequest, Uri}
import akka.stream.scaladsl.{Flow, Source}
import akka.stream.{...
0
votes
1
answer
78
views
Validating akka-stream Source in unit test
What is idiomatic way of validating a akka-stream Source in a unit test?
I am doing like this:
f.service.fetchData(id).flatMap {
case Right(source) => {
// TODO: I need to ...
1
vote
0
answers
65
views
akka-stream and mockito: value thenReturn is not a member of Nothing
I am trying to mock a method which returns type:
Future[Option[(ContentType, Long, Source[ByteString, NotUsed])]]
I mock as:
when(service.fetchData(id)).thenReturn(Future.successful(
Some(...
0
votes
0
answers
61
views
Akka stream: unexpected completed tcp stream
I've crated a two-way tcp stream using akka-stream-typed.
void createSocket(ActorSystem<Void> actorSystem, ActorRef<SomeApi> ownerRef, String serverIp, int serverPort) {
Tcp.get(...
0
votes
1
answer
64
views
Fetching data from S3 with akka-streams, akka-http
I am trying to read a file from S3 and send it back to client from a akka-http based rest endpoint, like this:
def fetchFromS3(s3Uri: String):
Future[Either[CustomError, Source[ByteString, ...
0
votes
1
answer
105
views
How to unmarshall json response removing unnecessary fields using Akka HTTP
I'm new to Akka HTTP and I want to get rid of unnecessary fields from a JSON response and take only the necessary fields. For example, I use this endpoint to get the response and it contains a bunch ...
0
votes
1
answer
46
views
akka-http - Multipart/FromData stream batch size
I'm receiving a Multipart/FromData as a server. This request contains Source[Multipart.FormData.BodyPart, Any] and each Multipart.FormData.BodyPart contains Source[ByteString, Any] inside. I'm able to ...
0
votes
1
answer
59
views
Akka - convert Flow into Collection or Publisher
I'm trying to split an Akka Source into two separate ones.
val requestFlow = Flow[BodyPartEntity].to(Sink.seq) // convert to Seq[BodyPartEntity]
val dataFlow = Flow[BodyPartEntity].to(Sink....
0
votes
0
answers
42
views
Akka - multipart with metadata and huge data
I'm receiving the following multipart data:
{
"meta": "filename",
"data": "data bytes" // about 10gb
}
And I have a service that saves data bytes into DB:
...
1
vote
1
answer
63
views
Akka Streams: How to form inlets and outlets for a Graph using a Flow
I have some code that is similar to the following:
object Test extends App {
val SomeComplicatedFlow: Flow[Int, Int, NotUsed] =
Flow.fromGraph(GraphDSL.create() { implicit builder =>
...
0
votes
2
answers
60
views
How to handle backpressure when Streaming file from s3 with actor interop
I am trying to download a large file from S3 and sending it's data to another actor that is doing an http request and then to persist the response. I want to limit number of requests sent by that ...
0
votes
1
answer
63
views
Akka flow Input (`In`) as Output (`Out`)
I am trying to write a piece of code which does following:-
Reads a large csv file from remote source like s3.
Process the file record by record.
Send notification to user
Write the output to a ...