Updating to use avro_turf and configuration settings for Environment …

…variables

field_struct_avro_schema.gemspec

@@ -39,6 +39,7 @@ Gem::Specification.new do |spec|
   spec.add_dependency 'activemodel'
   spec.add_dependency 'avro', '~> 1.11.0'
   spec.add_dependency 'avro-builder'
+  spec.add_dependency 'avro_turf'
   spec.add_dependency 'excon'
 
   spec.add_development_dependency 'bundler'

lib/field_struct/avro_schema.rb

@@ -7,6 +7,8 @@
 
 require 'avro_acima'
 require 'avro/builder'
+require 'avro_turf'
+require 'avro_turf/confluent_schema_registry'
 require 'excon'
 require 'field_struct'
 

lib/field_struct/avro_schema/configuration.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+require 'uri'
+
+module FieldStruct
+  module AvroSchema
+    class Configuration
+      attr_accessor :user_name, :password, :schema_registry_url, :automatic_schema_registration
+
+      def initialize
+        @user_name = env_or_default('KAFKA_SCHEMA_REGISTRY_USERNAME', nil)
+        @password = env_or_default('KAFKA_SCHEMA_REGISTRY_PASSWORD', nil)
+        @schema_registry_url = env_or_default('KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8081')
+        @automatic_schema_registration = env_or_default('KAFKA_AUTO_REGISTER_SCHEMAS', 'false') == 'true'
+      end
+
+      def schema_registry_base_url
+        uri = URI(schema_registry_url)
+        uri.path = ''
+        uri.query = nil
+        uri.to_s
+      end
+
+      def schema_registry_path_prefix
+        uri = URI(schema_registry_url)
+        uri.path
+      end
+
+      private
+
+      def env_or_default(name, default_value)
+        value = ENV.fetch(name, nil)
+        value = default_value if value.blank?
+        value
+      end
+    end
+  end
+end

lib/field_struct/avro_schema/kafka.rb

@@ -1,7 +1,6 @@
 # frozen_string_literal: true
 
 require_relative 'kafka/in_memory_cache'
-require_relative 'kafka/schema_registry'
 require_relative 'kafka/cached_schema_registry'
 require_relative 'kafka/schema_store'
 
@@ -13,6 +12,7 @@
 require_relative 'kafka/coders/json_encoder'
 require_relative 'kafka/coders/string_decoder'
 require_relative 'kafka/coders/string_encoder'
+require_relative 'configuration'
 
 module FieldStruct
   module AvroSchema
@@ -34,6 +34,10 @@ def logger=(value)
         @logger = value
       end
 
+      def configuration
+        @configuration ||= Configuration.new
+      end
+
       def events
         @events ||= {}
       end
@@ -43,7 +47,12 @@ def schema_registry
       end
 
       def base_schema_registry
-        @base_schema_registry ||= SchemaRegistry.new registry_url, logger: logger
+        @base_schema_registry ||=
+          AvroTurf::ConfluentSchemaRegistry.new configuration.schema_registry_base_url,
+                                                user: configuration.user_name,
+                                                password: configuration.password,
+                                                path_prefix: configuration.schema_registry_path_prefix,
+                                                logger: logger
       end
 
       def registry_url
@@ -99,11 +108,14 @@ def register_event_schemas
       def register_event_schema(klass)
         return nil unless klass.publishable?
 
-        id = if ENV.fetch('KAFKA_AUTO_REGISTER', 'false') == 'true'
+        id = if configuration.automatic_schema_registration
                schema_registry.register build_subject_name(klass), klass.schema
              else
-               schema_registry.check build_subject_name(klass), klass.schema
+               data = schema_registry.check build_subject_name(klass), klass.schema
+               data.fetch('id')
              end
+        raise StandardError, "Schema Not Found -- Schema Name: #{klass.default_schema_record_name}" if id.blank?
+
         klass.schema_id id
         klass
       rescue StandardError => e

lib/field_struct/avro_schema/kafka/cached_schema_registry.rb

@@ -71,10 +71,16 @@ def mock!
         def to_s
           format '#<%s upstream=%s cache=%s>',
                  self.class.name,
-                 @upstream.inspect,
+                 upstream_inspect,
                  @cache.inspect
         end
 
+        def upstream_inspect
+          format '#<%s url=%s>',
+                 @upstream.class.name,
+                 @upstream.instance_variable_get(:@url).inspect
+        end
+
         alias inspect to_s
       end
     end

spec/kafka_spec.rb

@@ -45,15 +45,14 @@
     it { expect(subject).to be_a FieldStruct::AvroSchema::Kafka::CachedSchemaRegistry }
     it do
       expect(subject.inspect).to eq '#<FieldStruct::AvroSchema::Kafka::CachedSchemaRegistry ' \
-                                    'upstream=#<FieldStruct::AvroSchema::Kafka::SchemaRegistry ' \
+                                    'upstream=#<AvroTurf::ConfluentSchemaRegistry ' \
                                     'url=nil> cache=#<FieldStruct::AvroSchema::Kafka::InMemoryCache ' \
                                     'schemas_by_id=0 ids_by_schema=0 schema_by_subject_version=0>>'
     end
   end
   describe '.base_schema_registry' do
     subject { described_class.base_schema_registry }
-    it { expect(subject).to be_a FieldStruct::AvroSchema::Kafka::SchemaRegistry }
-    it { expect(subject.inspect).to eq '#<FieldStruct::AvroSchema::Kafka::SchemaRegistry url=nil>' }
+    it { expect(subject).to be_a AvroTurf::ConfluentSchemaRegistry }
   end
   describe '.registry_url' do
     subject { described_class.registry_url }

spec/structs/company_spec.rb

@@ -503,11 +503,11 @@
       it('to_hash') { compare instance.to_hash, company_attrs.deep_stringify_keys }
     end
   end
-  context 'registration', :vcr, :registers, :env_change do
-    let(:env) { { 'KAFKA_AUTO_REGISTER' => 'true' } }
+  context 'registration', :vcr, :registers do
     let(:registration) { kafka.register_event_schema described_class }
     it('Kafka has event registered') { expect(kafka.events[described_class.name]).to eq described_class }
     it 'registers with schema_registry' do
+      expect(kafka.configuration).to receive(:automatic_schema_registration).and_return(true)
       expect { registration }.to_not raise_error
       expect(described_class.schema_id).to eq exp_schema_id
     end

spec/structs/custom_record_name_spec.rb

@@ -189,11 +189,11 @@
       it('to_hash') { compare instance.to_hash, person_attrs.deep_stringify_keys }
     end
   end
-  context 'registration', :env_change do
-    let(:env) { { 'KAFKA_AUTO_REGISTER' => 'true' } }
+  context 'registration' do
     let(:registration) { kafka.register_event_schema described_class }
     it('Kafka has event registered') { expect(kafka.events[described_class.name]).to eq described_class }
     it 'registers with schema_registry', :vcr, :registers do
+      expect(kafka.configuration).to receive(:automatic_schema_registration).and_return(true)
       expect { registration }.to_not raise_error
       expect(described_class.schema_id).to eq exp_schema_id
     end

spec/structs/developer_spec.rb

@@ -217,11 +217,11 @@
       it('to_hash') { compare instance.to_hash, developer_attrs.deep_stringify_keys }
     end
   end
-  context 'registration', :env_change do
-    let(:env) { { 'KAFKA_AUTO_REGISTER' => 'true' } }
+  context 'registration' do
     let(:registration) { kafka.register_event_schema described_class }
     it('Kafka has event registered') { expect(kafka.events[described_class.name]).to eq described_class }
     it 'registers with schema_registry', :vcr, :registers do
+      expect(kafka.configuration).to receive(:automatic_schema_registration).and_return(true)
       expect { registration }.to_not raise_error
       expect(described_class.schema_id).to eq exp_schema_id
     end

spec/structs/employee_spec.rb

@@ -205,11 +205,11 @@
       it('to_hash') { compare instance.to_hash, employee_attrs.deep_stringify_keys }
     end
   end
-  context 'registration', :env_change do
-    let(:env) { { 'KAFKA_AUTO_REGISTER' => 'true' } }
+  context 'registration' do
     let(:registration) { kafka.register_event_schema described_class }
     it('Kafka has event registered') { expect(kafka.events[described_class.name]).to eq described_class }
     it 'registers with schema_registry', :vcr, :registers do
+      expect(kafka.configuration).to receive(:automatic_schema_registration).and_return(true)
       expect { registration }.to_not raise_error
       expect(described_class.schema_id).to eq exp_schema_id
     end

spec/structs/friend_spec.rb

@@ -247,11 +247,11 @@
       it('to_hash') { compare instance.to_hash, friend_attrs.deep_stringify_keys }
     end
   end
-  context 'registration', :env_change do
-    let(:env) { { 'KAFKA_AUTO_REGISTER' => 'true' } }
+  context 'registration' do
     let(:registration) { kafka.register_event_schema described_class }
     it('Kafka has event registered') { expect(kafka.events[described_class.name]).to eq described_class }
     it 'registers with schema_registry', :vcr, :registers do
+      expect(kafka.configuration).to receive(:automatic_schema_registration).and_return(true)
       expect { registration }.to_not raise_error
       expect(described_class.schema_id).to eq exp_schema_id
     end

spec/structs/person_spec.rb

@@ -189,11 +189,11 @@
       it('to_hash') { compare instance.to_hash, person_attrs.deep_stringify_keys }
     end
   end
-  context 'registration', :env_change do
-    let(:env) { { 'KAFKA_AUTO_REGISTER' => 'true' } }
+  context 'registration' do
     let(:registration) { kafka.register_event_schema described_class }
     it('Kafka has event registered') { expect(kafka.events[described_class.name]).to eq described_class }
     it 'registers with schema_registry', :vcr, :registers do
+      expect(kafka.configuration).to receive(:automatic_schema_registration).and_return(true)
       expect { registration }.to_not raise_error
       expect(described_class.schema_id).to eq exp_schema_id
     end

spec/structs/runner_spec.rb

@@ -262,11 +262,11 @@
       it('to_hash') { compare instance.to_hash, runner_attrs.deep_stringify_keys }
     end
   end
-  context 'registration', :env_change do
-    let(:env) { { 'KAFKA_AUTO_REGISTER' => 'true' } }
+  context 'registration' do
     let(:registration) { kafka.register_event_schema described_class }
     it('Kafka has event registered') { expect(kafka.events[described_class.name]).to eq described_class }
     it 'registers with schema_registry', :vcr, :registers do
+      expect(kafka.configuration).to receive(:automatic_schema_registration).and_return(true)
       expect { registration }.to_not raise_error
       expect(described_class.schema_id).to eq exp_schema_id
     end