Questions tagged [alpakka]
Alpakka is the collective name for various Akka Streams connectors, integration patterns, and data transformations.
204
questions
0
votes
2
answers
40
views
Scala adding an element to JsValue before convertT[class]
I have a Kafka consumed record that will be parsed as JsValue with spray.json in scala, but I also have some data in the record's header, and I want to do:
Consume record with Alpakka Kafka library (...
0
votes
0
answers
42
views
Unable to put item in dynalite table using testcontainer
I am trying to test Dynamodb operations using Dynalite test containers. I am able to create and list the table but it's failing with the below error while trying to put items in the table.
Caused by: ...
0
votes
0
answers
10
views
Sharing the KafkaProducer instance for alpakka kafka client
From the alpakka-kafka documentation, I am a bit confused reading about sharing a Kafka producer instance. In the doc, it first says,
The underlying KafkaProducer is thread-safe and sharing a single ...
0
votes
1
answer
54
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
34
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
0
answers
37
views
Alpakka S3 library for "assume role"
I am trying to push records to AWS S3 with Aplakka S3 library. The issue is due to security issues, I have to "assume role", and my IAM user doesn't have access to PUT. with aws cli, I could ...
0
votes
0
answers
58
views
How to configure Alpakka Kafka Consumers depending on Partition Number and Akka Nodes
I have a question about Alpakka Kafka behaviour by handling partitions for consumers.
For ex, how Alpakka Kafka will act in the following configuration.
Let's say, I have a topicA with 4 partitions ...
0
votes
0
answers
48
views
Kafka does not reconsume messages after group-id has changed and offset.reset is set to earliest
I am trying to test a feature that requires reconsuming the kafka messages from an earlier offset. I changed the group id into a new one and the auto.offset.reset = "earliest", but that didn'...
0
votes
1
answer
36
views
How to log Kafka Producer Config on Startup
I currently have an application with inbound and outbound of kafka, however upon launch I can see the Consumer config being logged as so:
INFO [o.a.k.c.c.ConsumerConfig] - ConsumerConfig values:
...
0
votes
1
answer
46
views
Influxdb Alpakka connector not writing to database
I'm trying to write into the Influxdb (running in a docker container with version 2.0).
I'm using Scala and Reactive Streams. Therefore the Alpakka connector (https://doc.akka.io/docs/alpakka/current/...
2
votes
0
answers
86
views
alpakka kafka auto offset reset config latest does not read messages in test case
I have a java play application that leverages akka streams to read from kafka and output to a websocket. my application creates a kafka source using the following:
Consumer.plainSource(...
0
votes
0
answers
25
views
Kafka alPakka throwing compile issues when using restart source
Trying to use restart with source in case of Kafka rebalance , however below code is throwing error when using Alpakka Kafka library
Throwing complies error while using below code
Caused by: ...
0
votes
0
answers
25
views
Configure List of S3 Attributes to read/write in multiple S3 accounts
Got a use case where I have to do read and write operations to multiple object store which are S3 implementations(Ceph and AWS S3). I am using alpakka and added both config for ceph and s3 in app....
0
votes
0
answers
84
views
How getOffsetsOnAssign works?
How actually getOffsetsOnAssign in Consumer.plainPartitionedManualOffsetSource (this mean we keep offset externally) works?
Will consumer always wait for getOffsetOnAssign or it has a timeout?
It ...
0
votes
1
answer
75
views
Alpakka Kafka Java Test
I am trying to write Junit tests for an app that uses kafka streams to emit data from kafka to a websocket connection. I have been able to run the application locally publishing to the topic and ...
1
vote
0
answers
41
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....
1
vote
1
answer
113
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
0
answers
78
views
How to distribute workload to many compute and do scatter-gather scenarios with Kafka Steam?
I am new to Kafka Stream and Alpakka Kafka.
Problem: I have been using Java Executor Service to run parallel jobs and when ALL of them are done, marking the entire process done. The issue is fault ...
1
vote
0
answers
99
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
97
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 ...
0
votes
0
answers
84
views
AWS S3 and Ceph Object Store simultaneous read and write using Alpakka S3 client
I am using alpakka S3 client to RW data to Ceph Object Storage. Here is the following config:
alpakka.s3{
aws {
credentials {
provider = static
access-key-id = <...
0
votes
1
answer
88
views
Scala Generic Repository Class For Reactive Mongo Repository(alpakka) - Needed Class Found T
Im trying to create a Generic Class in Scala so I can create a repository for different collection without repeating myself.
The problem is that if I do it as a Generic Class(as in this example) I get ...
0
votes
2
answers
66
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
73
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 ...
3
votes
0
answers
74
views
Does Kafka support creating projections from multiple category streams?
We are using EventStore for our event sourced solution. We would like to export events into Kafka and do stream processing using Kafka. In EventStore each type of stream belongs to a category, for ...
0
votes
2
answers
121
views
How do you transform a `FixedSqlAction` into a `StreamingDBIO` in Slick?
I'm creating an akka-stream using Alpakka and the Slick module but I'm stuck in a type mismatch problem.
One branch is about getting the total number of invoices in their table:
def getTotal(...
0
votes
0
answers
52
views
How to to stream a directory of parquet files through akka/alpakka streams
I trying to stream a directory of parquet files through akka/alpakka streams, the problem is although I am setting the directory as the source, it streams only one file.
Any advice will be greatly ...
1
vote
1
answer
59
views
How to map parquet record (json) to case class using alpakka
I have the data that saved in parquet file as json e.g
{"name":"john", "age":23.5}
I want to convert it to
case class Person(name: String, age: Double) so I can use ...
0
votes
2
answers
92
views
akka streaming file lines to actor router and writing with single actor. how to handle the backpressure
I want to stream a file from s3 to actor to be parsed and enriched and to write the output to other file.
The number of parserActors should be limited e.g
application.conf
akka{
actor{
...
1
vote
1
answer
429
views
Can Throttle rate of consuming incoming messages using Akka Kafka stream
Is there a way to read only X number of message in particular time period suppose 1 minutes using Akka kafka Stream Consumer https://doc.akka.io/docs/alpakka-kafka/0.15/consumer.html via some ...
0
votes
0
answers
104
views
akka streams kafka consumer stops in few seconds
I am reading from kafka consumer, then persist message to db and send to producer on different topic. My akka streams app stops running in few seconds after launch.
Here is how my stream looks like.
...
0
votes
1
answer
192
views
how to initialize a continous running stream using alpakka, spring boot & Akka-stream?
All,
I am developing an application, which use alpakka spring boot integration to read data from kafka. I have most of the code ready, the only place i am stuck is how to initialize a continuous ...
0
votes
1
answer
130
views
Kafka consumers (same group-id) stucked in reading always from same partition
I have 2 consumers running with the same group-id and reading from topic having 3 partititons and parsing messages with KafkaAvroDeserializer. The consumer has these settings:
def ...
0
votes
1
answer
242
views
How to read parquet file from S3 using akka streams or alpakka
Im trying to read parque file from S3 using akka streams following the official doc but I am getting this error
java.io.IOException: No FileSystem for scheme: s3a
this is the code that triggered that ...
1
vote
1
answer
87
views
Combine prefixAndTail(1) with Sink.lazySink for SubFlow created by .splitAfter
I am currently developing an akka-stream/alpakka application that has the following general logic
Given a Flow, split it into a SubFlow using the splitAfter method.
For each of the SubFlow's in point ...
0
votes
1
answer
101
views
Not able to read data from elasticsearch using alpakka
I was trying to read document(json data stored in ES) from elasticsearch using alpakka.
I got this alpakka-Elasticsearch.
Here it says that you can stream messages from or to Elasticsearch using the
...
0
votes
1
answer
51
views
No implicits found for parameter sprayJsonReader: JsonReader[T]
I have created an sbt project to learn how to use elastic search with akka. I came across alpakka which provides this feature (to connect with elasticsearch).
According to docs, to search from ES we ...
0
votes
0
answers
56
views
Slick Query with calling a procedure does not terminate the program
I have a Main program that queries data from DB using slick. It constructs JSON from the record it gets from DB and then publishes to some kafka topics. Almost at the end of the program, I m trying to ...
0
votes
1
answer
230
views
Alpakka and Akka Version Compatibility Issue
When I try to sbt build my project, the project build fails with an "extracting product structure failed" error. I am suspecting something related to the versions I am using for Alpakka and ...
0
votes
1
answer
292
views
Adjusting parallism based on number of partitions assigned in Consumer.committablePartitionedSource
I am trying to use Consumer.committablePartitionedSource() and creating stream per partition as shown below
public void setup() {
control = Consumer.committablePartitionedSource(...
1
vote
2
answers
358
views
Can MQTT v3 client work with MQTT v5 server?
I wanted to make use of Shared Subscription feature of MQTT v5. But currently I am using AKKA MQTT client which doesn't support MQTT v5 client. Can I still use v3 MQTT paho client and use Shared ...
2
votes
0
answers
35
views
presigned url for aws s3 in akka java application
I am using Akka framework and writing a Java application that talks to AWS S3. Is there any example of generating a pre-signed get request for s3 in java using Akka framework? Alpakka S3Client does ...
1
vote
0
answers
217
views
Akka streams: Alpakka not using more than one CPU core
We have created a Alpakka stream, which consumes Kafka message from a topic and then process those messages. These messages are processed in parallel, using mapAsyncUnordered with a configured ...
1
vote
0
answers
344
views
Inclusion of .futureValue method in scala.concurrent.Future?
I am working through a tutorial on getting data from an S3 bucket via this tutorial (the full test code can be found here).
When I use the code
val s3names: Source[Option[(Source[ByteString, NotUsed], ...
1
vote
0
answers
111
views
Number of Messages on akka-streams Backpressure
As well-explained here thanks to Sean Glover, I have been exploring how backpressure works on Alpakka.
As I see, when mailbox size is not defined, the default mailbox size is set to be unbounded.
My ...
0
votes
1
answer
198
views
MongoDB Reactive Streams run-time dependency error with Alpakka Mongo Connector ClassNotFoundException
I'm trying to integrate the Alpakka Mongo Connector into an application that heavily relies on the Akka libraries for stream processing. The application utilizes Akka HTTP as well.
I am encountering a ...
0
votes
1
answer
130
views
How to download from Google Cloud Storage by Alpakka-gcs without providing secret-key?
I'm using Alpakka-gcs connecting to GCS from google-compute-engine perfectly if I provide gcs-secret-key on application.conf like the below.
alpakka.google.cloud.storage {
project-id = "...
0
votes
1
answer
182
views
Why does my Alpakka SFTP connector never connect?
I'm new to Alpakka/Akka Streams and I'm trying to set up a stream where I stream data between two SFTP servers with my system in the middle, here's the code.
import akka.actor.ActorSystem
import akka....
0
votes
1
answer
202
views
Reactive akka stream : How to delay the graph shutdown until the source has dried out?
I am trying to feed data (say 10 strings) to an akka-stream graph from a kafka broker
Here's the graph in an unittest :
val consumer = AlpakkaConsumer(KafkaBootstrapServer(kafkaURL).value, "...
0
votes
1
answer
161
views
can't get akka streams / alpakka S3 to work in simple case
I'm doing what I think is a very simple thing to check that alpakka is working:
val awsCreds = AwsBasicCredentials.create("xxx", "xxx")
val credentialsProvider = ...