Add s3 backup #8

mdedetrich · 2021-07-22T15:58:58Z

This is an in progress PR for backing up into Amazon S3. Some important notes

You cannot rename an object in S3 (apart from just copying it to another key which is inappropriate for the large files we will be creating), see https://stackoverflow.com/a/26820022. This means the only way to identify if a file is still being uploaded is by using specific S3 SDK's that allow you to list all multipart uploads in progress (unfortunately this is not part of the Alpakka API). Good news is that the API allows you to resume from a previously terminated multi upload so we don't have to re-download everything, we can just resume.
Figuring out how to deal with committing ReducedConsumerRecord cursors is quite complicated. This is because you have a cursor for each ReducedConsumerRecord however when we are feeding the Flow into the S3.multipartUploadWithHeaders Sink we lose the concept of single elements of ReducedConsumerRecord with a Context (since at the point of S3.multipartUploadWithHeaders its only dealing with a stream of bytes with no beginning or end).
It is a possibility that when dividing the period into the time elapsed from the start of the stream that we can overflow. This can be alleviated by using a different ChronoUnit (currently we use MICROS). A lazy way of fixing this issue is to use BigInt rather than Long (which can basically increase until you run out of memory. There may be a smarter way of handling this problem by detecting that if we are reaching Long.MaxValue we just reset the counter but doing so will make the logic less simple.

jlprat

Added all the feedback I could think off.
Overall seems to be good. But obviously without tests it's complicated.

core-backup/src/main/scala/aiven/io/guardian/kafka/backup/BackupClientInterface.scala

core-backup/src/main/scala/aiven/io/guardian/kafka/backup/Config.scala

jlprat · 2021-07-27T12:24:58Z

core-backup/src/main/scala/aiven/io/guardian/kafka/backup/BackupClientInterface.scala

+sealed abstract class BackupStreamPosition
+
+object BackupStreamPosition {
+
+  /** The backup stream has just started right now
+    */
+  case object Start extends BackupStreamPosition
+
+  /** The backup stream is in the middle of a time period
+    */
+  case object Middle extends BackupStreamPosition
+
+  /** The backup stream position has just hit a boundary for when a new period starts
+    */
+  case object Boundary extends BackupStreamPosition
+}


If those are meant to be public, I would put them in their own file.
I personally prefer having 1 class per file, easier to navigate and locate things.

I may make this private because I don't think it needs to be public, but I note your point (I will put it in a different file if it ends up being public)

core-backup/src/main/scala/aiven/io/guardian/kafka/backup/BackupClientInterface.scala

core-s3/src/main/scala/aiven/io/guardian/kafka/s3/configs/S3.scala

core/src/main/scala/aiven/io/guardian/kafka/models/ReducedConsumerRecord.scala

jlprat

Looks good.
Would it be possible to use new commits instead of force pushing? Reviewing is easier with newer commits

mdedetrich · 2021-07-28T11:08:30Z

@jlprat Okay sure will do, this will be last time I will force push. Do not that github still keeps conversations on comments even if you have force pushed, it only gets hidden if the actual code gets changed.

jlprat · 2021-07-28T11:27:09Z

LGTM. Let me know when it's not a draft anymore

mdedetrich · 2021-08-05T12:32:16Z

core-backup/src/main/scala/aiven/io/guardian/kafka/backup/BackupClientInterface.scala

@@ -117,7 +117,7 @@ trait BackupClientInterface {
              (reducedConsumerRecord, period)
            }

-          case None => throw new IllegalAccessException("")
+          case None => throw Errors.ExpectedStartOfSource


Via testing I have verified that this case isn't possible, you have to have at least a single element for a Source in order for it to even start

mdedetrich · 2021-08-05T12:32:20Z

core-backup/src/main/scala/aiven/io/guardian/kafka/backup/BackupClientInterface.scala

-            // TODO Is it possible to hit this branch? I assume if the Stream is started its impossible for
-            // head to be empty
-            ???
+          case None => throw Errors.ExpectedStartOfSource


Via testing I have verified that this case isn't possible, you have to have at least a single element for a Source in order for it to even start

mdedetrich · 2021-08-05T12:32:55Z

core-backup/src/main/scala/aiven/io/guardian/kafka/backup/BackupClientInterface.scala

@@ -204,5 +228,5 @@ object BackupClientInterface {
                                                      reducedConsumerRecord: ReducedConsumerRecord
  ): Long =
    // TODO handle overflow?
-    ChronoUnit.MICROS.between(reducedConsumerRecord.toOffsetDateTime, initialTime) / period.toMicros
+    ChronoUnit.MICROS.between(initialTime, reducedConsumerRecord.toOffsetDateTime) / period.toMicros


One bug that I found in initial implementation, turns out you can negative time between 2 time periods!

Don't really get this change

jlprat

I have some comments mostly on the testing area.

jlprat · 2021-08-10T09:25:22Z

core-backup/src/test/scala/aiven/io/guardian/kafka/backup/BackupClientInterfaceSpec.scala

+import scala.annotation.nowarn
+import scala.concurrent.Await
+import scala.concurrent.duration.{FiniteDuration, _}
+import scala.language.postfixOps


This import is not needed

jlprat · 2021-08-10T11:37:06Z

core-backup/src/test/scala/aiven/io/guardian/kafka/backup/BackupClientInterfaceSpec.scala

+  "BackupClientInterface" can {
+    "splitAtBoundaryCondition" should {
+      "BackupStreamPosition.Boundary happy case" in {


I don't know if it improved or not, but nested matchers were bad at performance. A big flat spec was several times faster than a nested one.

jlprat · 2021-08-10T11:39:58Z

core-backup/src/test/scala/aiven/io/guardian/kafka/backup/BackupClientInterfaceSpec.scala

+    "calculateBackupStreamPositions" should {
+
+      "must always have at least one BackupStreamPosition.Boundary" in {


Nit, either use must instead of should or remove it from the string

jlprat · 2021-08-10T11:41:42Z

core-backup/src/test/scala/aiven/io/guardian/kafka/backup/BackupClientInterfaceSpec.scala

+        }
+      }
+
+      "Every ReducedConsumerRecord after a BackupStreamPosition.Boundary must be in the next consecutive time period" in {


That sentence makes no sense:
"calculateBackupStreamPositions" should Every ReducedConsumerRecord after a BackupStreamPosition.Boundary must be in the next consecutive time period

jlprat · 2021-08-10T11:42:20Z

core-backup/src/test/scala/aiven/io/guardian/kafka/backup/BackupClientInterfaceSpec.scala

+        }
+      }
+
+      "The time difference between two consecutive BackupStreamPosition.Middle has to be less then the time period" in {


Same here. Probably means the test belongs somewhere else, or you can structure the nesting differently

jlprat · 2021-08-10T11:47:49Z

core/src/test/scala/aiven/io/guardian/kafka/ScalaTestConstants.scala

+import scala.language.postfixOps
+
+trait ScalaTestConstants {
+  val AwaitTimeout = 10 minutes


That's huge. Probably seconds might be enough. What about 5 seconds?

jlprat · 2021-08-10T11:48:41Z

core-backup/src/main/scala/aiven/io/guardian/kafka/backup/BackupClientInterface.scala

@@ -204,5 +228,5 @@ object BackupClientInterface {
                                                      reducedConsumerRecord: ReducedConsumerRecord
  ): Long =
    // TODO handle overflow?
-    ChronoUnit.MICROS.between(reducedConsumerRecord.toOffsetDateTime, initialTime) / period.toMicros
+    ChronoUnit.MICROS.between(initialTime, reducedConsumerRecord.toOffsetDateTime) / period.toMicros


Don't really get this change

mdedetrich · 2021-08-10T11:53:54Z

@jlprat

Don't really get this change

In regards to ChronoUnit.MICROS.between, one gives positive values and the other gives negatives.

mdedetrich marked this pull request as draft July 22, 2021 15:59

mdedetrich force-pushed the add-s3-backup branch 14 times, most recently from 2b3cdb5 to f1fd7dd Compare July 27, 2021 10:55

jlprat reviewed Jul 27, 2021

View reviewed changes

mdedetrich force-pushed the add-s3-backup branch from f1fd7dd to 329124d Compare July 28, 2021 10:51

jlprat reviewed Jul 28, 2021

View reviewed changes

mdedetrich force-pushed the add-s3-backup branch from 329124d to 3d4bdde Compare July 28, 2021 11:07

mdedetrich force-pushed the add-s3-backup branch from 3d4bdde to ad88508 Compare July 28, 2021 11:08

mdedetrich force-pushed the add-s3-backup branch from ad88508 to e3b59b6 Compare July 30, 2021 08:30

mdedetrich added 2 commits August 5, 2021 02:02

Add backup Config

d6397b7

Core implementation of backup into S3

5071d25

mdedetrich force-pushed the add-s3-backup branch from e3b59b6 to 5071d25 Compare August 5, 2021 00:02

mdedetrich commented Aug 5, 2021

View reviewed changes

mdedetrich force-pushed the add-s3-backup branch from 9e44344 to 65842fe Compare August 5, 2021 14:57

mdedetrich added 2 commits August 5, 2021 16:58

Add in tests for processing the kafka stream into time slices

967c577

Added in unexpected error for the head of a Source

606c048

mdedetrich force-pushed the add-s3-backup branch from 65842fe to 606c048 Compare August 5, 2021 14:59

Cleanup and refactor tests

896075f

mdedetrich marked this pull request as ready for review August 6, 2021 09:05

mdedetrich merged commit 783bf9c into main Aug 9, 2021

mdedetrich deleted the add-s3-backup branch August 9, 2021 09:25

jlprat reviewed Aug 10, 2021

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add s3 backup #8

Add s3 backup #8

mdedetrich commented Jul 22, 2021 •

edited

Loading

jlprat left a comment

jlprat Jul 27, 2021

mdedetrich Jul 27, 2021

jlprat left a comment

mdedetrich commented Jul 28, 2021 •

edited

Loading

jlprat commented Jul 28, 2021

mdedetrich Aug 5, 2021

mdedetrich Aug 5, 2021

mdedetrich Aug 5, 2021

jlprat Aug 10, 2021

jlprat left a comment

jlprat Aug 10, 2021

jlprat Aug 10, 2021

jlprat Aug 10, 2021

jlprat Aug 10, 2021

jlprat Aug 10, 2021

jlprat Aug 10, 2021

jlprat Aug 10, 2021

mdedetrich commented Aug 10, 2021 •

edited

Loading

		"calculateBackupStreamPositions" should {

		"must always have at least one BackupStreamPosition.Boundary" in {

Add s3 backup #8

Add s3 backup #8

Conversation

mdedetrich commented Jul 22, 2021 • edited Loading

jlprat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jlprat left a comment

Choose a reason for hiding this comment

mdedetrich commented Jul 28, 2021 • edited Loading

jlprat commented Jul 28, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jlprat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mdedetrich commented Aug 10, 2021 • edited Loading

mdedetrich commented Jul 22, 2021 •

edited

Loading

mdedetrich commented Jul 28, 2021 •

edited

Loading

mdedetrich commented Aug 10, 2021 •

edited

Loading