Add Kinesis batch publisher with retries #408

aserrallerios · 2017-07-16T11:21:14Z

This adds a Kinesis publisher with batching and retries. It does not ensure publish order (due to batching and retries; I plan to add more publishers that ensure total ordering or ordering by partition key).

More tests and documentation will be added and the stage logic may be refactored/simplified.

Feedback will be much appreciated!

raboof · 2017-07-17T13:41:47Z

Travis reports 4 compile errors when building with Scala 2.11 - might be good to look into?

aserrallerios · 2017-07-17T14:25:01Z

Fixed :)

iamthiago · 2017-07-19T21:19:49Z

@aserrallerios Today I had to create something very similar to be used in my pipeline, unfortunately, only the Source part is present on the master branch, so I have created this:

https://gist.github.com/thiagoandrade6/4fb8684bb2409995f781151c42e349ac

It is completely based on the SNS and Lambda sub projects. IMHO perhaps you should adapt your code to look like that, but as I said, it's only my opinion :)

aserrallerios · 2017-07-19T21:30:22Z

Or instead you can review the PR and ask for changes if you think something is missing/wrong :)

Your Flow lacks parallelism and retries, both implemented in this PR. This PR also includes throttling and other configuration parameters.

iamthiago · 2017-07-19T21:41:47Z

Don't worry, I just like the way SNS and Lambda implement their GraphStages.

But indeed, my flow is just simple enough to run in my application today, not really production ready for everybody, I just wanted to show because I couldn't wait for your PR to be approved :)

aserrallerios · 2017-07-21T08:57:02Z

Yeah, but adding features to a stage makes it more complex and not quite as comparable.

If you cannot wait, please review :), also @raboof and @areyouspiffy feedback would be great.

I'll try to add the missing tests/docs as soon as possible.

brianmowen · 2017-07-24T15:47:29Z

kinesis/src/main/scala/akka/stream/alpakka/kinesis/KinesisFlowErrors.scala

+
+import scala.util.control.NoStackTrace
+
+object KinesisFlowErrors {


If we renamed this KinesisErrors I could also put KinesisSourceErrors in here as well, but it's up to you if you think they're better off in separate files. Same with settings.

Done.

About the settings, I'd rather have them separated, it could be a mess mixing all those companion objects, validations and constructors. And think that we may have more Kinesis Sources/Flows in the future with different settings.

brianmowen · 2017-07-24T15:47:45Z

👍 Good work!

aserrallerios · 2017-08-01T07:49:34Z

I think it's feature-complete, ready to be reviewed. Feel free to ask for changes or new stuff.

nvrs · 2017-08-23T06:49:35Z

+1 for this, good work.

raboof · 2017-08-23T08:45:24Z

kinesis/src/main/scala/akka/stream/alpakka/kinesis/KinesisFlowStage.scala

+      }
+
+      private def checkForCompletion() =
+        if (inFlight <= 0 && pendingRequests.isEmpty && waitingRetries.isEmpty && isClosed(in)) {


should inFlight ever be able to become < 0? Perhaps log a warning then?

No, it cannot happen. I'll change it to == 0.

raboof · 2017-08-23T08:46:16Z

docs/src/main/paradox/kinesis.md

@@ -60,10 +60,10 @@ Java
 The `KinesisSource` creates one `GraphStage` per shard. Reading from a shard requires an instance of `ShardSettings`. 

 Scala
-: @@snip (../../../../kinesis/src/test/scala/akka/stream/alpakka/kinesis/scaladsl/Examples.scala) { #settings }
+: @@snip (../../../../kinesis/src/test/scala/akka/stream/alpakka/kinesis/scaladsl/Examples.scala) { #source-settings }


Would be nice to add the root of the project as a variable like we do for akka, but that's for another PR

raboof

Generally LGTM, including tests and documentation. Haven't worked with Kinesis myself though so can't really comment on specifics.

aserrallerios · 2017-08-27T09:55:20Z

Fixed-up commits.

eduardojld · 2017-08-29T19:40:02Z

@aserrallerios I took the liberty of reviewing your code, and it looks great, in general there is not much difference from what I was doing, but there is something I would not do, and its the retries, I do not think it should be a responsibility of the flow. You even put in some nice things like batching and throttling, So Im not sure would be a good decision to make retries part of the flow.

Other than that I would just add a KinesisSink to complete the whole picture like in SQS

aserrallerios · 2017-08-29T21:08:08Z

Thnx for the review @eduardojld. I'll add a Sink version of the publish.

About the retires, note that the code doesn't retry any "internal" or "unknown" server error, just partial errors i.e. records that failed to publish among other records that were successfully published. For example, if you PUT 500 records, the first 100 may succeed and the remaining 400 may fail. We have found that handling this partials errors is mandatory if you really need to get your PUT throughput near the available throughput limit (depending on the number of shards), as they are quite common. Removing the retries logic makes it nearly unusable if high throughput is required.

An alternative to internal retries would be to publish failed records downstream, forcing the user to decide what to do with failed records. But honestly I cannot think in any other scenario than the user wanting to feed the same "publish records flow" again, to retry those records, producing a cycle in the graph. I'd rather do it internally unless the community find strong arguments for it, as it would increase the complexity of the end-user graphs (with graph cycles!).

You can always disable retries with settings parameter :D

aserrallerios · 2017-09-06T09:57:40Z

Added sink version.

aserrallerios · 2017-10-03T09:16:05Z

Guys, more reviews, anything before accepting it?

etspaceman · 2017-10-03T17:52:11Z

It looks like there's some conflicts w/ your branch @aserrallerios - Those likely have to be resolved prior to a merge.

aserrallerios · 2017-10-05T07:16:21Z

Thanks @etspaceman 😅

Cherry-picked

nvrs · 2017-10-24T09:06:52Z

Hi @aserrallerios, the merge of #535, which changed the formatting of many files including KinesisSourceErrors, is now causing a conflict since you have renamed that file to KinesisErrors.

Do you need any help with this? I am keen on getting this for the next release :)

aserrallerios · 2017-10-24T10:47:31Z

Rebased & formatted, thanks @nvrs!

nvrs · 2017-10-25T13:06:37Z

Hi @raboof , do you think this needs more reviewers before accepting the PR?

aserrallerios · 2017-10-25T15:28:11Z

Guys, I've been testing with Graphs that self-restart on failures, and I have one question: do the initialization code of the StageLogic execute on each restart of the Stage or just the code on preStart method? (for example when using akka's RestartSource.withBackoff).

If the answer is that only the code on preStart executes, I'll need to initialize all variables of the StageLogic inside the preStart method, to avoid malfunction of the Stage after Graph restarts.

aserrallerios · 2017-10-26T07:17:54Z

It didn't harm, anyway, so I've done it: change initialization of Stage's variables & some minor stuff to improve code clearness in the Stage.

raboof added the p:kinesis label Jul 17, 2017

aserrallerios mentioned this pull request Jul 19, 2017

Kinesis publisher Flow/Sink #382

Closed

brianmowen reviewed Jul 24, 2017

View reviewed changes

raboof reviewed Aug 23, 2017

View reviewed changes

raboof approved these changes Aug 23, 2017

View reviewed changes

aserrallerios force-pushed the kinesis-publisher branch from 91c06a0 to 4440968 Compare August 27, 2017 09:53

aserrallerios mentioned this pull request Aug 29, 2017

Kinesis Support for KPL and KCL Libraries #433

Closed

aserrallerios force-pushed the kinesis-publisher branch from 4440968 to 89ca54c Compare September 6, 2017 09:54

etspaceman mentioned this pull request Sep 30, 2017

Support integration with AWS Kinesis using the Message Broker API lagom/lagom#839

Open

aserrallerios force-pushed the kinesis-publisher branch from 89ca54c to 912fb2c Compare October 5, 2017 06:40

aserrallerios mentioned this pull request Oct 10, 2017

Add Kinesis KCL Source #434

Closed

Add Kinesis batch publisher with retries

faeb774

aserrallerios added 2 commits October 24, 2017 12:41

Add Sink version of Kinesis publisher

8defeba

Format source

9a667f8

aserrallerios force-pushed the kinesis-publisher branch from 912fb2c to 9a667f8 Compare October 24, 2017 10:47

aserrallerios force-pushed the kinesis-publisher branch 2 times, most recently from 3629d6d to 218f7ff Compare October 26, 2017 09:20

Initialize Stage safely & clarify Stage logic

79c2654

aserrallerios force-pushed the kinesis-publisher branch from 218f7ff to 79c2654 Compare October 26, 2017 12:17

raboof merged commit 6cff13a into akka:master Oct 30, 2017

aserrallerios deleted the kinesis-publisher branch April 26, 2019 09:56

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Kinesis batch publisher with retries #408

Add Kinesis batch publisher with retries #408

aserrallerios commented Jul 16, 2017

raboof commented Jul 17, 2017

aserrallerios commented Jul 17, 2017

iamthiago commented Jul 19, 2017

aserrallerios commented Jul 19, 2017 •

edited

Loading

iamthiago commented Jul 19, 2017

aserrallerios commented Jul 21, 2017

brianmowen Jul 24, 2017

aserrallerios Jul 25, 2017

brianmowen commented Jul 24, 2017

aserrallerios commented Aug 1, 2017

nvrs commented Aug 23, 2017

raboof Aug 23, 2017

aserrallerios Aug 27, 2017

raboof Aug 23, 2017

raboof left a comment

aserrallerios commented Aug 27, 2017

eduardojld commented Aug 29, 2017

aserrallerios commented Aug 29, 2017 •

edited

Loading

aserrallerios commented Sep 6, 2017

aserrallerios commented Oct 3, 2017 •

edited

Loading

etspaceman commented Oct 3, 2017

aserrallerios commented Oct 5, 2017

nvrs commented Oct 24, 2017 •

edited

Loading

aserrallerios commented Oct 24, 2017

nvrs commented Oct 25, 2017

aserrallerios commented Oct 25, 2017 •

edited

Loading

aserrallerios commented Oct 26, 2017


		import scala.util.control.NoStackTrace

		object KinesisFlowErrors {

Add Kinesis batch publisher with retries #408

Add Kinesis batch publisher with retries #408

Conversation

aserrallerios commented Jul 16, 2017

raboof commented Jul 17, 2017

aserrallerios commented Jul 17, 2017

iamthiago commented Jul 19, 2017

aserrallerios commented Jul 19, 2017 • edited Loading

iamthiago commented Jul 19, 2017

aserrallerios commented Jul 21, 2017

brianmowen Jul 24, 2017

Choose a reason for hiding this comment

aserrallerios Jul 25, 2017

Choose a reason for hiding this comment

brianmowen commented Jul 24, 2017

aserrallerios commented Aug 1, 2017

nvrs commented Aug 23, 2017

raboof Aug 23, 2017

Choose a reason for hiding this comment

aserrallerios Aug 27, 2017

Choose a reason for hiding this comment

raboof Aug 23, 2017

Choose a reason for hiding this comment

raboof left a comment

Choose a reason for hiding this comment

aserrallerios commented Aug 27, 2017

eduardojld commented Aug 29, 2017

aserrallerios commented Aug 29, 2017 • edited Loading

aserrallerios commented Sep 6, 2017

aserrallerios commented Oct 3, 2017 • edited Loading

etspaceman commented Oct 3, 2017

aserrallerios commented Oct 5, 2017

nvrs commented Oct 24, 2017 • edited Loading

aserrallerios commented Oct 24, 2017

nvrs commented Oct 25, 2017

aserrallerios commented Oct 25, 2017 • edited Loading

aserrallerios commented Oct 26, 2017

aserrallerios commented Jul 19, 2017 •

edited

Loading

aserrallerios commented Aug 29, 2017 •

edited

Loading

aserrallerios commented Oct 3, 2017 •

edited

Loading

nvrs commented Oct 24, 2017 •

edited

Loading

aserrallerios commented Oct 25, 2017 •

edited

Loading