chore: Feature/kafka consumer

.github/workflows/codeql-analysis.yml

@@ -39,6 +39,9 @@ jobs:
       with:
         languages: ${{ matrix.language }}
 
+    - name: Install global tools
+      run: dotnet tool install --global Apache.Avro.Tools
+
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: Autobuild

.github/workflows/examples-tests.yml

@@ -33,6 +33,9 @@ jobs:
       - name: Install dependencies
         run: dotnet restore
 
+      - name: Install global tools
+        run: dotnet tool install --global Apache.Avro.Tools
+
       - name: Build
         run: dotnet build --configuration Release --no-restore /tl
 

examples/Kafka/Avro/src/Avro.csproj

@@ -0,0 +1,35 @@
+<Project Sdk="Microsoft.NET.Sdk">
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+        <GenerateRuntimeConfigurationFiles>true</GenerateRuntimeConfigurationFiles>
+        <AWSProjectType>Lambda</AWSProjectType>
+        <!-- This property makes the build directory similar to a publish directory and helps the AWS .NET Lambda Mock Test Tool find project dependencies. -->
+        <CopyLocalLockFileAssemblies>true</CopyLocalLockFileAssemblies>
+        <!-- Generate ready to run images during publishing to improve cold start time. -->
+        <PublishReadyToRun>true</PublishReadyToRun>
+
+        <AssemblyName>Avro.Example</AssemblyName>
+
+    </PropertyGroup>
+    <ItemGroup>
+        <PackageReference Include="Amazon.Lambda.RuntimeSupport" Version="1.12.0"/>
+        <PackageReference Include="Amazon.Lambda.Core" Version="2.5.0"/>
+        <PackageReference Include="Amazon.Lambda.Serialization.SystemTextJson" Version="2.4.4"/>
+        <PackageReference Include="AWS.Lambda.Powertools.Logging" Version="2.0.0" />
+    </ItemGroup>
+    <ItemGroup>
+        <ProjectReference Include="..\..\..\..\libraries\src\AWS.Lambda.Powertools.Kafka.Avro\AWS.Lambda.Powertools.Kafka.Avro.csproj" />
+    </ItemGroup>
+    <Target Name="GenerateAvroClasses" BeforeTargets="CoreCompile">
+        <Exec Command="avrogen -s $(ProjectDir)CustomerProfile.avsc $(ProjectDir)Generated"/>
+    </Target>
+    <ItemGroup>
+        <None Remove="kafka-avro-event.json" />
+        <Content Include="kafka-avro-event.json">
+            <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+        </Content>
+    </ItemGroup>
+</Project>

examples/Kafka/Avro/src/CustomerProfile.avsc

@@ -0,0 +1,46 @@
+{
+  "type": "record",
+  "name": "CustomerProfile",
+  "namespace": "com.example",
+  "fields": [
+    {"name": "user_id", "type": "string"},
+    {"name": "full_name", "type": "string"},
+    {"name": "email", "type": {
+      "type": "record",
+      "name": "EmailAddress",
+      "fields": [
+        {"name": "address", "type": "string"},
+        {"name": "verified", "type": "boolean"},
+        {"name": "primary", "type": "boolean"}
+      ]
+    }},
+    {"name": "age", "type": "int"},
+    {"name": "address", "type": {
+      "type": "record",
+      "name": "Address",
+      "fields": [
+        {"name": "street", "type": "string"},
+        {"name": "city", "type": "string"},
+        {"name": "state", "type": "string"},
+        {"name": "country", "type": "string"},
+        {"name": "zip_code", "type": "string"}
+      ]
+    }},
+    {"name": "phone_numbers", "type": {
+      "type": "array",
+      "items": {
+        "type": "record",
+        "name": "PhoneNumber",
+        "fields": [
+          {"name": "number", "type": "string"},
+          {"name": "type", "type": {"type": "enum", "name": "PhoneType", "symbols": ["HOME", "WORK", "MOBILE"]}}
+        ]
+      }
+    }},
+    {"name": "preferences", "type": {
+      "type": "map",
+      "values": "string"
+    }},
+    {"name": "account_status", "type": {"type": "enum", "name": "AccountStatus", "symbols": ["ACTIVE", "INACTIVE", "SUSPENDED"]}}
+  ]
+}

examples/Kafka/Avro/src/Function.cs

@@ -0,0 +1,21 @@
+using Amazon.Lambda.Core;
+using Amazon.Lambda.RuntimeSupport;
+using AWS.Lambda.Powertools.Kafka;
+using AWS.Lambda.Powertools.Kafka.Avro;
+using AWS.Lambda.Powertools.Logging;
+using com.example;
+
+string Handler(ConsumerRecords<string, CustomerProfile> records, ILambdaContext context)
+{
+    foreach (var record in records)
+    {
+        Logger.LogInformation("Record Value: {@record}", record.Value);
+    }
+
+    return "Processed " + records.Count() + " records";
+}
+
+await LambdaBootstrapBuilder.Create((Func<ConsumerRecords<string, CustomerProfile>, ILambdaContext, string>?)Handler,
+        new PowertoolsKafkaAvroSerializer()) // Use PowertoolsKafkaAvroSerializer for Avro serialization
+    .Build()
+    .RunAsync();

examples/Kafka/Avro/src/Generated/com/example/AccountStatus.cs

@@ -0,0 +1,23 @@
+// ------------------------------------------------------------------------------
+// <auto-generated>
+//    Generated by avrogen, version 1.12.0+8c27801dc8d42ccc00997f25c0b8f45f8d4a233e
+//    Changes to this file may cause incorrect behavior and will be lost if code
+//    is regenerated
+// </auto-generated>
+// ------------------------------------------------------------------------------
+namespace com.example
+{
+	using System;
+	using System.Collections.Generic;
+	using System.Text;
+	using global::Avro;
+	using global::Avro.Specific;
+
+	[global::System.CodeDom.Compiler.GeneratedCodeAttribute("avrogen", "1.12.0+8c27801dc8d42ccc00997f25c0b8f45f8d4a233e")]
+	public enum AccountStatus
+	{
+		ACTIVE,
+		INACTIVE,
+		SUSPENDED,
+	}
+}

examples/Kafka/Avro/src/Generated/com/example/Address.cs

@@ -0,0 +1,115 @@
+// ------------------------------------------------------------------------------
+// <auto-generated>
+//    Generated by avrogen, version 1.12.0+8c27801dc8d42ccc00997f25c0b8f45f8d4a233e
+//    Changes to this file may cause incorrect behavior and will be lost if code
+//    is regenerated
+// </auto-generated>
+// ------------------------------------------------------------------------------
+namespace com.example
+{
+	using System;
+	using System.Collections.Generic;
+	using System.Text;
+	using global::Avro;
+	using global::Avro.Specific;
+
+	[global::System.CodeDom.Compiler.GeneratedCodeAttribute("avrogen", "1.12.0+8c27801dc8d42ccc00997f25c0b8f45f8d4a233e")]
+	public partial class Address : global::Avro.Specific.ISpecificRecord
+	{
+		public static global::Avro.Schema _SCHEMA = global::Avro.Schema.Parse("{\"type\":\"record\",\"name\":\"Address\",\"namespace\":\"com.example\",\"fields\":[{\"name\":\"st" +
+				"reet\",\"type\":\"string\"},{\"name\":\"city\",\"type\":\"string\"},{\"name\":\"state\",\"type\":\"s" +
+				"tring\"},{\"name\":\"country\",\"type\":\"string\"},{\"name\":\"zip_code\",\"type\":\"string\"}]}" +
+				"");
+		private string _street;
+		private string _city;
+		private string _state;
+		private string _country;
+		private string _zip_code;
+		public virtual global::Avro.Schema Schema
+		{
+			get
+			{
+				return Address._SCHEMA;
+			}
+		}
+		public string street
+		{
+			get
+			{
+				return this._street;
+			}
+			set
+			{
+				this._street = value;
+			}
+		}
+		public string city
+		{
+			get
+			{
+				return this._city;
+			}
+			set
+			{
+				this._city = value;
+			}
+		}
+		public string state
+		{
+			get
+			{
+				return this._state;
+			}
+			set
+			{
+				this._state = value;
+			}
+		}
+		public string country
+		{
+			get
+			{
+				return this._country;
+			}
+			set
+			{
+				this._country = value;
+			}
+		}
+		public string zip_code
+		{
+			get
+			{
+				return this._zip_code;
+			}
+			set
+			{
+				this._zip_code = value;
+			}
+		}
+		public virtual object Get(int fieldPos)
+		{
+			switch (fieldPos)
+			{
+			case 0: return this.street;
+			case 1: return this.city;
+			case 2: return this.state;
+			case 3: return this.country;
+			case 4: return this.zip_code;
+			default: throw new global::Avro.AvroRuntimeException("Bad index " + fieldPos + " in Get()");
+			};
+		}
+		public virtual void Put(int fieldPos, object fieldValue)
+		{
+			switch (fieldPos)
+			{
+			case 0: this.street = (System.String)fieldValue; break;
+			case 1: this.city = (System.String)fieldValue; break;
+			case 2: this.state = (System.String)fieldValue; break;
+			case 3: this.country = (System.String)fieldValue; break;
+			case 4: this.zip_code = (System.String)fieldValue; break;
+			default: throw new global::Avro.AvroRuntimeException("Bad index " + fieldPos + " in Put()");
+			};
+		}
+	}
+}