RIG Fails when KAFKA_SCHEMA_REGISTRY_HOST not set and not running on localhost

[chadmott]

Summary and context

IF KAFKA_SCHEMA_REGISTRY_HOST is not set, RIG defaults to localhost if there is no schema registry running on localhost, RIG fails to proxy requests and restarts itself

How to reproduce

When I run RIG with these settings:

Run RIG w/o a schema registry

If you point to valid schema registry, it works.

Describe how your clients are set up, whether they connect via SSE or WS, etc.

..I see the following error and/or log output:

Add the log output to help us determine your problem. If possible, reproduce the problem after setting the environment variable LOG_LEVEL to debug.

** (exit) an exception was raised:
    ** (Protocol.UndefinedError) protocol String.Chars not implemented for {%MatchError{term: {:error, %HTTPoison.Error{id: nil, reason: :econnrefused}}}, [{RigKafka.Avro, :__fetch_schema_memoize, 2, [file: 'lib/rig_kafka/avro.ex', line: 80]}, {Memoize.Cache, :do_get_or_run, 3, [file: 'lib/memoize/cache.ex', line: 107]}, {RigKafka.Avro, :decode, 2, [file: 'lib/rig_kafka/avro.ex', line: 20]}, {RigKafka.Client.GroupSubscriber, :handle_message, 4, [file: 'lib/rig_kafka/client.ex', line: 84]}, {:brod_group_subscriber, :handle_messages, 4, [file: 'src/brod_group_subscriber.erl', line: 540]}, {:brod_group_subscriber, :handle_info, 2, [file: 'src/brod_group_subscriber.erl', line: 333]}, {:gen_server, :try_dispatch, 4, [file: 'gen_server.erl', line: 637]}, {:gen_server, :handle_msg, 6, [file: 'gen_server.erl', line: 711]}]}. This protocol is implemented for: Atom, BitString, Date, DateTime, Float, Integer, List, NaiveDateTime, Rig.Subscription, Time, URI, Version, Version.Requirement
        (elixir) /usr/local/src/elixir/lib/elixir/lib/string/chars.ex:3: String.Chars.impl_for!/1
        (elixir) /usr/local/src/elixir/lib/elixir/lib/string/chars.ex:22: String.Chars.to_string/1
        (logger) lib/logger/formatter.ex:166: anonymous fn/1 in Logger.Formatter.output/5
        (elixir) lib/enum.ex:1314: Enum."-map/2-lists^map/1-0-"/2
        (elixir) lib/enum.ex:1314: Enum."-map/2-lists^map/1-0-"/2
        (logger) lib/logger/formatter.ex:152: anonymous fn/6 in Logger.Formatter.format/5
        (elixir) lib/enum.ex:1925: Enum."-reduce/3-lists^foldl/2-0-"/3
        (logger) lib/logger/formatter.ex:151: Logger.Formatter.format/5

A clear and concise description of what you expected to happen.

Versions (please complete the following information):**

• Host OS: RIG docker linux
• Frontend node supertest
• RIG version as shown on startup Master

[mmacai]

Thanks for the report, I'll check. Regarding the fact not using registry, is there any situation when Avro would run registry-less? I assume you can create your own registry and not use the Confluent one, just curious if you would ever use it completely without registry.

[chadmott]

this particular issue occurs without any other AVRO config... meaning, the topic RIG connects to is not avro encoded.

to answer the question around using AVRO w/o a registry... yes you can do it... just import the avro file locally. given that the whole point of the encoding is to ensure type safety across microservices, it doesnt make a lot of sense to do that tho

[kevinbader]

@chadmott Cannot reproduce, but perhaps I'm misunderstand the report. I'm doing this:

1. start Kafka (no Schema Registry running)
2. run RIG with KAFKA_BROKERS=localhost:9092 and LOG_LEVEL=debug as the only environment variables

I don't see the error you've posted above. What other environment variables do you set?

[mmacai]

Tested couple of scenarios and everything worked well, besides one. When Avro (or potentially any other serializer) is turned off, but you send binary message to RIG. It still tries to pattern match it as a binary. Which is ok, but then ends in Serializer code -- so tries to use also schemes which he can't obviously find (or connect to). So we should probably update apps/rig_kafka/lib/rig_kafka/client.ex:81 or apps/rig_kafka/lib/rig_kafka/client.ex:42 to check if serializer is turned "on" or perhaps if there is connection to schema registry.

@kevinbader what do you think?

[kevinbader]

@mmacai I think that when there's an Avro message but there's no serializer defined, we should produce an error message, but the Kafka handler probably shouldn't die (I think that's what the OP expects). What about this (not tested):

diff --git a/apps/rig_kafka/lib/rig_kafka/client.ex b/apps/rig_kafka/lib/rig_kafka/client.ex
index f1d8576..e57ee27 100644
--- a/apps/rig_kafka/lib/rig_kafka/client.ex
+++ b/apps/rig_kafka/lib/rig_kafka/client.ex
@@ -76,30 +76,37 @@ defmodule RigKafka.Client do
 
       content_type = get_content_type(headers_no_prefix, body)
 
-      decoded_body =
-        cond do
-          content_type == "avro/binary" ->
-            data =
+      try do
+        decoded_body =
+          case content_type do
+            "avro/binary" ->
+              data =
+                body
+                |> Serializer.decode_body!("avro", schema_registry_host: schema_registry_host)
+                |> Jason.decode!()
+
+              Map.merge(headers_no_prefix, %{data: data})
+
+            "application/json" ->
               body
-              |> Serializer.decode_body("avro", schema_registry_host)
-              |> Jason.decode!()
 
-            Map.merge(headers_no_prefix, %{data: data})
+            _ ->
+              body
+          end
 
-          content_type == "application/json" ->
-            body
+        case callback.(decoded_body) do
+          :ok ->
+            {:ok, :ack, state}
 
-          true ->
-            body
+          err ->
+            info = %{error: err, topic: topic, partition: partition, offset: offset}
+            Logger.error("Callback failed to handle message: #{inspect(info)}")
+            {:ok, :ack_no_commit, state}
         end
-
-      case callback.(decoded_body) do
-        :ok ->
-          {:ok, :ack, state}
-
+      rescue
         err ->
           info = %{error: err, topic: topic, partition: partition, offset: offset}
-          Logger.error("Callback failed to handle message: #{inspect(info)}")
+          Logger.error(fn -> {"failed to decode message: #{inspect(err)}", [info: info]} end)
           {:ok, :ack_no_commit, state}
       end
     end
diff --git a/apps/rig_kafka/lib/rig_kafka/serializer.ex b/apps/rig_kafka/lib/rig_kafka/serializer.ex
index 69c36ba..9c3fa25 100644
--- a/apps/rig_kafka/lib/rig_kafka/serializer.ex
+++ b/apps/rig_kafka/lib/rig_kafka/serializer.ex
@@ -11,12 +11,18 @@ defmodule RigKafka.Serializer do
 
   @prefix "ce-"
 
-  @spec decode_body(any(), encode_type) :: any()
-  def decode_body(body, "avro", schema_registry_host) do
-    Avro.decode(body, schema_registry_host)
+  # ---
+
+  @spec decode_body!(any(), encode_type, opts :: []) :: any()
+
+  def decode_body!(body, "avro", opts) do
+    case opts[:schema_registry_host] do
+      nil -> raise "cannot decode avro message: schema registry host not set"
+      schema_registry_host -> Avro.decode(body, schema_registry_host)
+    end
   end
 
-  def decode_body(body, nil), do: body
+  def decode_body!(body, nil, _), do: body
 
   # ---
 

[mmacai]

@kevinbader I had something similar on my mind, I'll test it today once I have few minutes.