Adding Avro support capability #199

samizzy · 2022-08-18T11:18:14Z

This PR is part of the effort to introduce Avro capabilities to Nakadi Clients.

This PR contains modifications that introduces Avro publishing and consumption capability, kept in mind to introduce no changes to existing users of the client.

gchudnov · 2022-08-19T08:29:04Z

@samizzy
Thank you for the PR.

It looks the build is failing. could you please check it?

and it might be worth to check the DCO section that required the commits to be signed off.

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

gchudnov · 2022-08-19T09:17:42Z

build.sbt

+    "ch.qos.logback"                   % "logback-classic"         % "1.2.7",
+    "org.apache.avro"                  % "avro"                    % "1.11.0",
+    "com.fasterxml.jackson.dataformat" % "jackson-dataformat-avro" % "2.13.3",
+    "com.fasterxml.jackson.module"    %% "jackson-module-scala"    % "2.13.2",


"com.fasterxml.jackson.dataformat" % "jackson-dataformat-avro" % "2.13.3", "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.13.2",

hm. given that the project is a kind of tied to circe, wondering if it worth to have new libraries for encoding decoding bundled-in.

As far as I know lots of users of this library was considering bundling circe as questionable. and here we increase coupling even more.

gchudnov · 2022-08-19T09:24:22Z

build.sbt

@@ -108,6 +111,8 @@ libraryDependencies ++= {
      Seq.empty
  })
 }
+// avro compiler features
+avroStringType := "String"


avroStringType := "String"

should it be always a string? Are there any docs to follow?

gchudnov · 2022-08-19T09:28:58Z

build.sbt

+    "com.iheart"                      %% "ficus"                   % "1.5.1",
+    "com.typesafe.scala-logging"      %% "scala-logging"           % "3.9.4",
+    "ch.qos.logback"                   % "logback-classic"         % "1.2.7",
+    "org.apache.avro"                  % "avro"                    % "1.11.0",


"org.apache.avro" % "avro" % "1.11.0",

given that in the documentation stated that there are plans to introduce more shema formats and protocols (e.g. protobuf) wondering if it is worth to split the project into several modules so that avro is not bundled-in?

Otherwise the project might become too coupled with dozens of other libraries.

gchudnov · 2022-08-19T10:10:57Z

src/main/avro/batch.consumption/batch.consumption.1.avsc

@@ -0,0 +1,49 @@
+{
+  "name": "ConsumptionBatch",


ConsumptionBatch

wondering, should it be SubscriptionEvent ?

gchudnov · 2022-08-19T10:18:51Z

src/main/avro/batch.publishing/batch.publishing.1.avsc

@@ -0,0 +1,143 @@
+{
+  "name": "PublishingBatch",


PublishingBatch

again wondering if it causes confusion with the batch name we had before avro-change?

gchudnov · 2022-08-19T12:06:19Z

src/main/scala/org/zalando/kanadi/api/Subscriptions.scala

    kanadiHttpConfig: HttpConfig,
    http: HttpExt,
    materializer: Materializer)
    extends SubscriptionsInterface {
  protected val logger: LoggerTakingImplicit[FlowId] = Logger.takingImplicit[FlowId](classOf[Subscriptions])
-  private val baseUri_                               = Uri(baseUri.toString)
+  protected val baseUri_                             = Uri(baseUri.toString)


protected

why do we need this change?

gchudnov · 2022-08-19T12:06:44Z

src/main/scala/org/zalando/kanadi/api/Subscriptions.scala

+                                   oAuth2TokenProvider,
+                                   AvroSchema(eventTypeName, etSchema.schema.asString.get, etSchema.version.get))))
+
+    Await.result(result, Duration.apply(5, TimeUnit.SECONDS)).getOrElse(throw SchemaNotFoundError(consumerSchema))


Await.result...

same comment as for the publishing case. it might be worth the user of the library to decide how to react, without introducing 5 sec parameter that might be not relevant for some of the clients at all.

gchudnov · 2022-08-19T12:08:28Z

src/main/scala/org/zalando/kanadi/models/GeneralError.scala

@@ -12,6 +12,10 @@ final case class OtherError(error: BasicServerError) extends Exception {
  override def getMessage: String = s"Error from server, response is $error"
 }

+final case class SchemaNotFoundError(schemaString: String) extends Exception {
+  override def getMessage: String = s"Schema with value $schemaString not found on server"


instead of overriding getMessage, can we just provide a parameter to Exception ?

gchudnov · 2022-08-19T12:09:21Z

src/main/scala/org/zalando/kanadi/models/GeneralError.scala

@@ -12,6 +12,10 @@ final case class OtherError(error: BasicServerError) extends Exception {
  override def getMessage: String = s"Error from server, response is $error"
 }

+final case class SchemaNotFoundError(schemaString: String) extends Exception {


and wondering what is the reason to use Exception here? Should it be RuntimeException or some other common error class, related to the nakadi library?

gchudnov · 2022-08-19T12:10:33Z

src/test/scala/org/zalando/kanadi/api/AvroBasicSpec.scala

@@ -0,0 +1,133 @@
+package org.zalando.kanadi.api


given the complexity of the added code, it might be worth to add more comprehensive tests, given that the library might be used in critical projects.

gchudnov · 2022-08-19T12:15:11Z

@samizzy
forgot to post a comment the review:
besides of the code changes, the documentation update is needed to make it clear how to use the new capabilities.

vadeg

Thanks for the contribution! I have left some comments as well but mostly I +1 whan @gchudnov already found. I want to highlight three main points from my side:

Code requires decomposition of avro code from non avro. Having everything in one place complicates the code.
Using Await in production code is unacceptable. This could lead to many problems and such implementation is not suitable for critical services.
Integration test should cover different failure scenarios. Not only successful publish and consumption.

vadeg · 2022-08-19T13:14:47Z

project/plugins.sbt

@@ -6,3 +6,6 @@ addSbtPlugin("org.scoverage"                     % "sbt-scoverage"      % "1.9.2
 addSbtPlugin("org.scoverage"                     % "sbt-coveralls"      % "1.3.1")
 addSbtPlugin("com.codecommit"                    % "sbt-github-actions" % "0.13.0")
 addSbtPlugin("com.github.sbt"                    % "sbt-release"        % "1.1.0")
+addSbtPlugin("com.github.sbt"                    % "sbt-avro"           % "3.4.0")
+
+libraryDependencies += "org.apache.avro" % "avro-compiler" % "1.11.0"


Is it possible to use the same avro version here and in build.sbt?

This has been resolved in 443366a

vadeg · 2022-08-19T13:15:56Z

src/main/avro/batch.consumption/batch.consumption.1.avsc

@@ -0,0 +1,49 @@
+{


Could you elaborate on file naming schema? What is 1 in the name? Is it required?

vadeg · 2022-08-19T13:16:15Z

src/main/avro/batch.consumption/batch.consumption.1.avsc

+        "null",
+        "string"
+      ],
+      "default": null


What is this field?

vadeg · 2022-08-19T13:18:58Z

src/main/scala/org/zalando/kanadi/api/EventTypes.scala

+      response <- http.singleRequest(request)
+      result <- {
+        if (response.status == StatusCodes.NotFound) {
+          response.discardEntityBytes()


Definitely worth it :)

vadeg · 2022-08-19T13:20:21Z

src/main/scala/org/zalando/kanadi/api/Events.scala

@@ -98,6 +107,8 @@ object Event {
      }
  }

+  final case class AvroEvent[T](override val data: T, metadata: Metadata) extends Event[T](data)
+


I guess it is here because of sealed above.

vadeg · 2022-08-19T14:06:31Z

src/main/scala/org/zalando/kanadi/api/Events.scala

+            "The http server closed the connection unexpectedly before delivering responses for") =>
+        retryUnexpectedFailure(events, count, e, currentDuration)
+      case httpServiceError: HttpServiceError
+          if httpServiceError.httpResponse.status.intValue().toString.startsWith("5") =>


or event make comparison with ServerError class.

vadeg · 2022-08-19T14:07:48Z

src/main/scala/org/zalando/kanadi/api/Events.scala

+
+    val envelopes = events.map { event =>
+      new Envelope(toNakadiMetadata(event.metadata, publisherSchema.schemaVersion),
+                   ByteBuffer.wrap(writer.writeValueAsBytes(event.data)))


What will happen it there will be an exception thrown?

vadeg · 2022-08-19T14:14:42Z

src/main/scala/org/zalando/kanadi/api/Subscriptions.scala

+            }
+            .via(parseUserPayloadFlow)
+            .map {
+              case Left(error)   => throw error


You can not just throw an error from a stream and return Future.successful below. This will not work and cause many unexpected side effects.

vadeg · 2022-08-19T14:15:24Z

src/main/scala/org/zalando/kanadi/api/Subscriptions.scala

+            case StatusCodes.NotFound =>
+              unmarshalStringOrProblem(response.entity.withContentType(ContentTypes.`application/json`)).map {
+                stringOrProblem =>
+                  throw Subscriptions.Errors.SubscriptionNotFound(request, response, stringOrProblem)


vadeg · 2022-08-19T14:18:33Z

src/test/scala/org/zalando/kanadi/api/AvroBasicSpec.scala

+  override def beforeAll = Await.result(eventsTypesClient.create(eventType), 10 seconds)
+
+  override def afterAll =
+    Await.result(eventsTypesClient.delete(eventTypeName), 10 seconds)


Such tests are not sustainable and will lead to false positives depending on the environment where they run.

gchudnov · 2022-08-30T10:51:14Z

btw, it seems that failing tests are related to this issue: adyach/nakadi-docker#5
as it was mentioned in the other PR.

gchudnov · 2022-09-09T12:43:58Z

@samizzy
given changes to the license akka/akka#31561, no new features will be added here.

Instead propose to add this either to update the existing library for ZIO or create a new one and implement avro there.

but the main issue with Nakadi testing remains: we need to resolve this issue: adyach/nakadi-docker#5

mdedetrich · 2023-08-16T09:06:32Z

@gchudnov @vadeg I have just added a merge commit onto @samizzy PR which brings the latest changes from kanadi master (including migration from Akka to Pekko) so its in a state where it can be worked on.

One additional thing that I had to add to the merge commit is having to manually convert between CharSequence and String. Not sure why this is the case (compiling with JDK 11) but I can look into it later.

mdedetrich · 2023-08-16T09:11:30Z

If this gets merged it should be noted that unlike the JSON format, the Avro parsing isn't streamed (should probably mention this in README.md). This is possible to do via jsurfer, i.e. https://github.com/wanglingsong/JsonSurfer#binaray-format-jackson-only and you can use Pekko's existing JsonFraming as inspiration how to create a Flow that properly streams Avro binary data

mdedetrich · 2023-08-16T09:26:50Z

So CI failed

[error] org.specs2.specification.core.FatalExecution: Error from server, response is Problem(None,Bad Request,400,Some(Illegal enum value: 'avro_schema'. Possible values: [json_schema] (through reference chain: org.zalando.nakadi.domain.EventTypeBase["schema"]->org.zalando.nakadi.domain.EventTypeSchemaBase["type"])),None) (package.scala:83)
[error] org.zalando.kanadi.api.package$.$anonfun$processNotSuccessful$1(package.scala:83)
[error] CAUSED BY
[error] org.zalando.kanadi.models.GeneralError: Error from server, response is Problem(None,Bad Request,400,Some(Illegal enum value: 'avro_schema'. Possible values: [json_schema] (through reference chain: org.zalando.nakadi.domain.EventTypeBase["schema"]->org.zalando.nakadi.domain.EventTypeSchemaBase["type"])),None) (package.scala:83)
[error] org.zalando.kanadi.api.package$.$anonfun$processNotSuccessful$1(package.scala:83)

I guess this is due to the nakadi docker image used in CI is an outdated one which doesn't support avro

gchudnov · 2023-08-16T12:57:10Z

@mdedetrich
Thank you for the updates!
no worries, it will take some time to polish and merge this PR

Signed-off-by: Matthew de Detrich <mdedetrich@gmail.com>

samizzy requested review from mdedetrich, itachi3, pascalh and gchudnov as code owners August 18, 2022 11:18

Samrat Saha added 7 commits August 19, 2022 10:43

add avro generation plugin, avro and jackson avro dep

17d09b3

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

introduce avro_schema and implement fetchMatchingSchema method

6f51280

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

add avro publishing capability

779a07c

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

add avro consumption capability

6cd4bd4

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

add publishing and consuming schema files

357af3f

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

add basic tests

098b545

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

remove print statements

af4cbba

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

samizzy force-pushed the avro_support branch from 7e07e05 to af4cbba Compare August 19, 2022 08:43

apply scalafmt

d02f811

Signed-off-by: Samrat Saha <samrat.saha@zalando.de>

samizzy force-pushed the avro_support branch from 2437f2d to d02f811 Compare August 19, 2022 09:05

gchudnov reviewed Aug 19, 2022

View reviewed changes

vadeg suggested changes Aug 19, 2022

View reviewed changes

gchudnov mentioned this pull request Aug 30, 2022

arm64 -> amd64 adyach/nakadi-docker#5

Closed

mdedetrich force-pushed the avro_support branch from 8cd2e05 to 0acd01e Compare August 16, 2023 09:16

mdedetrich force-pushed the avro_support branch 4 times, most recently from 1c37273 to d59147c Compare August 16, 2023 10:17

mdedetrich force-pushed the avro_support branch 2 times, most recently from ca8f9ea to 700de7d Compare August 16, 2023 14:25

mdedetrich added 2 commits August 16, 2023 16:26

Merge remote-tracking branch 'origin/master' into avro_support

443366a

Signed-off-by: Matthew de Detrich <mdedetrich@gmail.com>

Add new lines

a02d1d1

Signed-off-by: Matthew de Detrich <mdedetrich@gmail.com>

mdedetrich force-pushed the avro_support branch from ccd908c to a02d1d1 Compare August 16, 2023 14:26

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding Avro support capability #199

Adding Avro support capability #199

samizzy commented Aug 18, 2022 •

edited

gchudnov commented Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov Aug 19, 2022

gchudnov commented Aug 19, 2022

vadeg left a comment

vadeg Aug 19, 2022

mdedetrich Aug 16, 2023 •

edited

vadeg Aug 19, 2022

vadeg Aug 19, 2022

vadeg Aug 19, 2022

vadeg Aug 19, 2022

vadeg Aug 19, 2022

vadeg Aug 19, 2022

vadeg Aug 19, 2022

vadeg Aug 19, 2022

vadeg Aug 19, 2022

gchudnov commented Aug 30, 2022

gchudnov commented Sep 9, 2022 •

edited

mdedetrich commented Aug 16, 2023 •

edited

mdedetrich commented Aug 16, 2023 •

edited

mdedetrich commented Aug 16, 2023

gchudnov commented Aug 16, 2023

Adding Avro support capability #199

Are you sure you want to change the base?

Adding Avro support capability #199

Conversation

samizzy commented Aug 18, 2022 • edited

gchudnov commented Aug 19, 2022

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gchudnov commented Aug 19, 2022

vadeg left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mdedetrich Aug 16, 2023 • edited

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gchudnov commented Aug 30, 2022

gchudnov commented Sep 9, 2022 • edited

mdedetrich commented Aug 16, 2023 • edited

mdedetrich commented Aug 16, 2023 • edited

mdedetrich commented Aug 16, 2023

gchudnov commented Aug 16, 2023

samizzy commented Aug 18, 2022 •

edited

mdedetrich Aug 16, 2023 •

edited

gchudnov commented Sep 9, 2022 •

edited

mdedetrich commented Aug 16, 2023 •

edited

mdedetrich commented Aug 16, 2023 •

edited