Re: [akka-user] akka 2.4.10 - connection reset by peer when consuming stream with parallelism == 1

[pbishnu1995]

[error] java.net.SocketException: Connection reset
[error] at 
sun.reflect.GeneratedConstructorAccessor8.newInstance(Unknown Source)
[error] at 
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
[error] at 
java.lang.reflect.Constructor.newInstance(Constructor.java:423)
[error] at 
sun.net.www.protocol.http.HttpURLConnection$10.run(HttpURLConnection.java:1926)
[error] at 
sun.net.www.protocol.http.HttpURLConnection$10.run(HttpURLConnection.java:1921)
[error] at java.security.AccessController.doPrivileged(Native 
Method)
[error] at 
sun.net.www.protocol.http.HttpURLConnection.getChainedException(HttpURLConnection.java:1920)
[error] at 
sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1490)
[error] at 
sun.net.www.protocol.http.HttpURLConnection.access$200(HttpURLConnection.java:91)
[error] at 
sun.net.www.protocol.http.HttpURLConnection$9.run(HttpURLConnection.java:1466)
[error] at 
sun.net.www.protocol.http.HttpURLConnection$9.run(HttpURLConnection.java:1464)
[error] at java.security.AccessController.doPrivileged(Native 
Method)
[error] at 
java.security.AccessController.doPrivilegedWithCombiner(AccessController.java:782)
[error] at 
sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1463)
[error] at 
java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:480)
[error] at 
scalaj.http.HttpRequest.scalaj$http$HttpRequest$$doConnection(Http.scala:359)
[error] at scalaj.http.HttpRequest.exec(Http.scala:335)
[error] at scalaj.http.HttpRequest.asString(Http.scala:455)
[error] at 
Test.RequestSender1$$anonfun$main$1$$anonfun$apply$mcVI$sp$1$$anon$1.run(RequestSender.scala:23)

This is the error I am getting if I am sending 2000+ request using threads 
to my akka http api. Can you help me you I am getting connection  reset 
Error?


On Friday, 30 September 2016 20:30:53 UTC+5:30, Konrad Malawski wrote:

> That means the other side has closed the connection. It could be an idle 
> timeout. Are you sure data is consistently flowing, and not being idle for 
> minutes?
>
> On 30 Sep 2016 16:59, "Eric Torti"  
> wrote:
>
>> Hey guys,
>>
>> I am new to akka streams so I may be missing something big here. Using 
>> akka-stream, akka-http-core, akka-typed-experimental all 2.4.10 - scala 
>> 2.11.8.
>>
>> I am experiencing the `connection reset by peer` error on a stream that I 
>> have checked with `curl` to remain open far longer than the time it takes 
>> akka to crash.
>>
>> The stream is produced from json lines like so: 
>>
>> ```
>>
>> def stream(): Source[Development, Any] = {
>>
>>   def chunkConsumer(res: HttpResponse) = {
>> res.entity.dataBytes
>>   .via(Framing.delimiter(ByteString("\r\n"), 1048576, allowTruncation = 
>> true))
>>   .map[Development](bs => { parse(bs.utf8String).extract[Development] })
>>   }
>>
>>   val req = HttpRequest(method = HttpMethods.GET, uri = 
>> s"http://$endpoint:$port/developments/stream;)
>>   val res: Source[Development, Any] = 
>> Source.single(req).via(client).flatMapConcat(chunkConsumer)
>>
>>   res
>> }
>>
>> ``` 
>>
>> And consumed like so:
>>
>> ```
>>
>> def slowAsyncTransformation(d: Development) = {
>>   Future { Thread sleep 50; d }
>> }
>>
>> val parallelism = 2
>>
>> DevelopmentsAPI
>>   .stream()
>>   .mapAsync(parallelism)(slowAsyncTransformation(_))
>>   .runFold(0)((acc, d) => {
>> System.out.println("processing development " + d.id)
>> acc + 1
>>   })
>>   .onComplete {
>> case Success(s) => {
>>   System.out.println(s"processed $s developments")
>>   system.terminate()
>>   Await.result(system.whenTerminated, 5 seconds)
>>   System.exit(0)
>> }
>> case Failure(ex) => {
>>   System.err.println(s"Could not finish processing developments: $ex")
>>   system.terminate()
>>   Await.result(system.whenTerminated, 5 seconds)
>>   System.exit(1)
>> }
>>   }
>>
>> ```
>>
>>
>> I can consistently get this to work from end to end in 180 seconds if I have 
>> `parallelism` set to 2. 
>>
>>
>> But if I set `parallelism` to 1, it crashes in 35 seconds with `Could not 
>> finish processing developments: akka.stream.StreamTcpException: The 
>> connection closed with error: Connection reset by peer`
>>
>> I cannot make sense of why is the degree of parallelism being 1 would 
>> cause the peer to reset the connection. 
>>
>> Any help will be greatly appreciated.
>>
>> Thanks, 
>>
>> Eric  
>>
>> -- 
>> >> Read the docs: http://akka.io/docs/
>> >> Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>> >> Search the archives: https://groups.google.com/group/akka-user
>> --- 
>> You received this message because you are subscribed to the Google Groups 
>> "Akka User List" 

Re: [akka-user] akka 2.4.10 - connection reset by peer when consuming stream with parallelism == 1

[Eric Torti]

Thank you for your answer, Konrad. 

I am guessing this shouldn't be the issue, because I have consumed the same 
stream very slowly using curl --limit-rate 75k. And it took me over 5 
minutes to go from end to end. And it worked ell. 

But I am not sure if akka http handles the socket the same way as curl 
does. So you might be right. I will try to emulate that setting parallelism 
to 1 and tuning the sleep interval. 

On Friday, September 30, 2016 at 12:00:53 PM UTC-3, Konrad Malawski wrote:
>
> That means the other side has closed the connection. It could be an idle 
> timeout. Are you sure data is consistently flowing, and not being idle for 
> minutes?
>
> On 30 Sep 2016 16:59, "Eric Torti"  
> wrote:
>
>> Hey guys,
>>
>> I am new to akka streams so I may be missing something big here. Using 
>> akka-stream, akka-http-core, akka-typed-experimental all 2.4.10 - scala 
>> 2.11.8.
>>
>> I am experiencing the `connection reset by peer` error on a stream that I 
>> have checked with `curl` to remain open far longer than the time it takes 
>> akka to crash.
>>
>> The stream is produced from json lines like so: 
>>
>> ```
>>
>> def stream(): Source[Development, Any] = {
>>
>>   def chunkConsumer(res: HttpResponse) = {
>> res.entity.dataBytes
>>   .via(Framing.delimiter(ByteString("\r\n"), 1048576, allowTruncation = 
>> true))
>>   .map[Development](bs => { parse(bs.utf8String).extract[Development] })
>>   }
>>
>>   val req = HttpRequest(method = HttpMethods.GET, uri = 
>> s"http://$endpoint:$port/developments/stream;)
>>   val res: Source[Development, Any] = 
>> Source.single(req).via(client).flatMapConcat(chunkConsumer)
>>
>>   res
>> }
>>
>> ``` 
>>
>> And consumed like so:
>>
>> ```
>>
>> def slowAsyncTransformation(d: Development) = {
>>   Future { Thread sleep 50; d }
>> }
>>
>> val parallelism = 2
>>
>> DevelopmentsAPI
>>   .stream()
>>   .mapAsync(parallelism)(slowAsyncTransformation(_))
>>   .runFold(0)((acc, d) => {
>> System.out.println("processing development " + d.id)
>> acc + 1
>>   })
>>   .onComplete {
>> case Success(s) => {
>>   System.out.println(s"processed $s developments")
>>   system.terminate()
>>   Await.result(system.whenTerminated, 5 seconds)
>>   System.exit(0)
>> }
>> case Failure(ex) => {
>>   System.err.println(s"Could not finish processing developments: $ex")
>>   system.terminate()
>>   Await.result(system.whenTerminated, 5 seconds)
>>   System.exit(1)
>> }
>>   }
>>
>> ```
>>
>>
>> I can consistently get this to work from end to end in 180 seconds if I have 
>> `parallelism` set to 2. 
>>
>>
>> But if I set `parallelism` to 1, it crashes in 35 seconds with `Could not 
>> finish processing developments: akka.stream.StreamTcpException: The 
>> connection closed with error: Connection reset by peer`
>>
>> I cannot make sense of why is the degree of parallelism being 1 would 
>> cause the peer to reset the connection. 
>>
>> Any help will be greatly appreciated.
>>
>> Thanks, 
>>
>> Eric  
>>
>> -- 
>> >> Read the docs: http://akka.io/docs/
>> >> Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>> >> Search the archives: https://groups.google.com/group/akka-user
>> --- 
>> You received this message because you are subscribed to the Google Groups 
>> "Akka User List" group.
>> To unsubscribe from this group and stop receiving emails from it, send an 
>> email to akka-user+...@googlegroups.com .
>> To post to this group, send email to akka...@googlegroups.com 
>> .
>> Visit this group at https://groups.google.com/group/akka-user.
>> For more options, visit https://groups.google.com/d/optout.
>>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] akka 2.4.10 - connection reset by peer when consuming stream with parallelism == 1

[Konrad Malawski]

That means the other side has closed the connection. It could be an idle
timeout. Are you sure data is consistently flowing, and not being idle for
minutes?

On 30 Sep 2016 16:59, "Eric Torti"  wrote:

> Hey guys,
>
> I am new to akka streams so I may be missing something big here. Using
> akka-stream, akka-http-core, akka-typed-experimental all 2.4.10 - scala
> 2.11.8.
>
> I am experiencing the `connection reset by peer` error on a stream that I
> have checked with `curl` to remain open far longer than the time it takes
> akka to crash.
>
> The stream is produced from json lines like so:
>
> ```
>
> def stream(): Source[Development, Any] = {
>
>   def chunkConsumer(res: HttpResponse) = {
> res.entity.dataBytes
>   .via(Framing.delimiter(ByteString("\r\n"), 1048576, allowTruncation = 
> true))
>   .map[Development](bs => { parse(bs.utf8String).extract[Development] })
>   }
>
>   val req = HttpRequest(method = HttpMethods.GET, uri = 
> s"http://$endpoint:$port/developments/stream;)
>   val res: Source[Development, Any] = 
> Source.single(req).via(client).flatMapConcat(chunkConsumer)
>
>   res
> }
>
> ```
>
> And consumed like so:
>
> ```
>
> def slowAsyncTransformation(d: Development) = {
>   Future { Thread sleep 50; d }
> }
>
> val parallelism = 2
>
> DevelopmentsAPI
>   .stream()
>   .mapAsync(parallelism)(slowAsyncTransformation(_))
>   .runFold(0)((acc, d) => {
> System.out.println("processing development " + d.id)
> acc + 1
>   })
>   .onComplete {
> case Success(s) => {
>   System.out.println(s"processed $s developments")
>   system.terminate()
>   Await.result(system.whenTerminated, 5 seconds)
>   System.exit(0)
> }
> case Failure(ex) => {
>   System.err.println(s"Could not finish processing developments: $ex")
>   system.terminate()
>   Await.result(system.whenTerminated, 5 seconds)
>   System.exit(1)
> }
>   }
>
> ```
>
>
> I can consistently get this to work from end to end in 180 seconds if I have 
> `parallelism` set to 2.
>
>
> But if I set `parallelism` to 1, it crashes in 35 seconds with `Could not
> finish processing developments: akka.stream.StreamTcpException: The
> connection closed with error: Connection reset by peer`
>
> I cannot make sense of why is the degree of parallelism being 1 would
> cause the peer to reset the connection.
>
> Any help will be greatly appreciated.
>
> Thanks,
>
> Eric
>
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] akka 2.4.10 - connection reset by peer when consuming stream with parallelism == 1

[Eric Torti]

Hey guys,

I am new to akka streams so I may be missing something big here. Using 
akka-stream, akka-http-core, akka-typed-experimental all 2.4.10 - scala 
2.11.8.

I am experiencing the `connection reset by peer` error on a stream that I 
have checked with `curl` to remain open far longer than the time it takes 
akka to crash.

The stream is produced from json lines like so: 

```

def stream(): Source[Development, Any] = {

  def chunkConsumer(res: HttpResponse) = {
res.entity.dataBytes
  .via(Framing.delimiter(ByteString("\r\n"), 1048576, allowTruncation = 
true))
  .map[Development](bs => { parse(bs.utf8String).extract[Development] })
  }

  val req = HttpRequest(method = HttpMethods.GET, uri = 
s"http://$endpoint:$port/developments/stream;)
  val res: Source[Development, Any] = 
Source.single(req).via(client).flatMapConcat(chunkConsumer)

  res
}

``` 

And consumed like so:

```

def slowAsyncTransformation(d: Development) = {
  Future { Thread sleep 50; d }
}

val parallelism = 2

DevelopmentsAPI
  .stream()
  .mapAsync(parallelism)(slowAsyncTransformation(_))
  .runFold(0)((acc, d) => {
System.out.println("processing development " + d.id)
acc + 1
  })
  .onComplete {
case Success(s) => {
  System.out.println(s"processed $s developments")
  system.terminate()
  Await.result(system.whenTerminated, 5 seconds)
  System.exit(0)
}
case Failure(ex) => {
  System.err.println(s"Could not finish processing developments: $ex")
  system.terminate()
  Await.result(system.whenTerminated, 5 seconds)
  System.exit(1)
}
  }

```


I can consistently get this to work from end to end in 180 seconds if I have 
`parallelism` set to 2. 


But if I set `parallelism` to 1, it crashes in 35 seconds with `Could not 
finish processing developments: akka.stream.StreamTcpException: The 
connection closed with error: Connection reset by peer`

I cannot make sense of why is the degree of parallelism being 1 would cause 
the peer to reset the connection. 

Any help will be greatly appreciated.

Thanks, 

Eric  

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.