-
Notifications
You must be signed in to change notification settings - Fork 539
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add Aspire.Confluent.Kafka component
apply pr suggestions apply pr suggestions apply pr suggestions Sort ConfigurationSchema.json properties Update ConfigurationSchema.json using ConfigSchemaGenerator apply pr suggestions apply pr suggestions apply pr suggesstions apply pr suggesstions apply pr suggestions apply pr suggestions drop kafka sample from this repo apply pr suggestions
- Loading branch information
Showing
36 changed files
with
3,170 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| // Licensed to the .NET Foundation under one or more agreements. | ||
| // The .NET Foundation licenses this file to you under the MIT license. | ||
|
|
||
| using Aspire.Hosting.ApplicationModel; | ||
| using Aspire.Hosting.Publishing; | ||
|
|
||
| namespace Aspire.Hosting; | ||
|
|
||
| public static class KafkaBuilderExtensions | ||
| { | ||
| private const int KafkaBrokerPort = 9092; | ||
| /// <summary> | ||
| /// Adds a Kafka broker container to the application. | ||
| /// </summary> | ||
| /// <param name="builder">The <see cref="IDistributedApplicationBuilder"/>.</param> | ||
| /// <param name="name">The name of the resource. This name will be used as the connection string name when referenced in a dependency.</param> | ||
| /// <param name="port">The host port of Kafka broker.</param> | ||
| /// <returns>A reference to the <see cref="IResourceBuilder{KafkaContainerResource}"/></returns> | ||
| public static IResourceBuilder<KafkaContainerResource> AddKafkaContainer(this IDistributedApplicationBuilder builder, string name, int? port = null) | ||
| { | ||
| var kafka = new KafkaContainerResource(name); | ||
| return builder.AddResource(kafka) | ||
| .WithEndpoint(hostPort: port, containerPort: KafkaBrokerPort) | ||
| .WithAnnotation(new ContainerImageAnnotation { Image = "confluentinc/confluent-local", Tag = "latest" }) | ||
| .WithManifestPublishingCallback(context => WriteKafkaContainerToManifest(context, kafka)) | ||
| .WithEnvironment(context => ConfigureKafkaContainer(context, kafka)); | ||
|
|
||
| static void WriteKafkaContainerToManifest(ManifestPublishingContext context, KafkaContainerResource resource) | ||
| { | ||
| context.WriteContainer(resource); | ||
| context.Writer.WriteString("connectionString", $"{{{resource.Name}.bindings.tcp.host}}:{{{resource.Name}.bindings.tcp.port}}"); | ||
| } | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Adds a Kafka resource to the application. A container is used for local development. | ||
| /// </summary> | ||
| /// <param name="builder">The <see cref="IDistributedApplicationBuilder"/>.</param> | ||
| /// <param name="name">The name of the resource. This name will be used as the connection string name when referenced in a dependency</param> | ||
| /// <returns>A reference to the <see cref="IResourceBuilder{KafkaServerResource}"/>.</returns> | ||
| public static IResourceBuilder<KafkaServerResource> AddKafka(this IDistributedApplicationBuilder builder, string name) | ||
| { | ||
| var kafka = new KafkaServerResource(name); | ||
| return builder.AddResource(kafka) | ||
| .WithEndpoint(containerPort: KafkaBrokerPort) | ||
| .WithAnnotation(new ContainerImageAnnotation{ Image = "confluentinc/confluent-local", Tag = "latest" }) | ||
| .WithManifestPublishingCallback(WriteKafkaServerToManifest) | ||
| .WithEnvironment(context => ConfigureKafkaContainer(context, kafka)); | ||
|
|
||
| static void WriteKafkaServerToManifest(ManifestPublishingContext context) | ||
| { | ||
| context.Writer.WriteString("type", "kafka.server.v0"); | ||
| } | ||
| } | ||
|
|
||
| private static void ConfigureKafkaContainer(EnvironmentCallbackContext context, IResource resource) | ||
| { | ||
| // confluentinc/confluent-local is a docker image that contains a Kafka broker started with KRaft to avoid pulling a separate image for ZooKeeper. | ||
| // See https://github.com/confluentinc/kafka-images/blob/master/local/README.md. | ||
| // When not explicitly set default configuration is applied. | ||
| // See https://github.com/confluentinc/kafka-images/blob/master/local/include/etc/confluent/docker/configureDefaults for more details. | ||
|
|
||
| var hostPort = context.PublisherName == "manifest" | ||
| ? KafkaBrokerPort | ||
| : GetResourcePort(resource); | ||
| context.EnvironmentVariables.Add("KAFKA_ADVERTISED_LISTENERS", | ||
| $"PLAINTEXT://localhost:29092,PLAINTEXT_HOST://localhost:{hostPort}"); | ||
|
|
||
| static int GetResourcePort(IResource resource) | ||
| { | ||
| if (!resource.TryGetAllocatedEndPoints(out var allocatedEndpoints)) | ||
| { | ||
| throw new DistributedApplicationException( | ||
| $"Kafka resource \"{resource.Name}\" does not have endpoint annotation."); | ||
| } | ||
|
|
||
| return allocatedEndpoints.Single().Port; | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| // Licensed to the .NET Foundation under one or more agreements. | ||
| // The .NET Foundation licenses this file to you under the MIT license. | ||
|
|
||
| using Aspire.Hosting.ApplicationModel; | ||
|
|
||
| namespace Aspire.Hosting; | ||
|
|
||
| /// <summary> | ||
| /// A resource that represents a Kafka broker container. | ||
| /// </summary> | ||
| /// <param name="name"></param> | ||
| public class KafkaContainerResource(string name) : ContainerResource(name), IResourceWithConnectionString, IResourceWithEnvironment | ||
| { | ||
| /// <summary> | ||
| /// Gets the connection string for Kafka broker. | ||
| /// </summary> | ||
| /// <returns>A connection string for the Kafka in the form "host:port" to be passed as <see href="https://docs.confluent.io/platform/current/clients/confluent-kafka-dotnet/_site/api/Confluent.Kafka.ClientConfig.html#Confluent_Kafka_ClientConfig_BootstrapServers">BootstrapServers</see>.</returns> | ||
| public string? GetConnectionString() | ||
| { | ||
| if (!this.TryGetAllocatedEndPoints(out var allocatedEndpoints)) | ||
| { | ||
| throw new DistributedApplicationException($"Kafka resource \"{Name}\" does not have endpoint annotation."); | ||
| } | ||
|
|
||
| return allocatedEndpoints.SingleOrDefault()?.EndPointString; | ||
| } | ||
|
|
||
| internal int GetPort() | ||
| { | ||
| if (!this.TryGetAllocatedEndPoints(out var allocatedEndpoints)) | ||
| { | ||
| throw new DistributedApplicationException($"Kafka resource \"{Name}\" does not have endpoint annotation."); | ||
| } | ||
|
|
||
| return allocatedEndpoints.Single().Port; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| // Licensed to the .NET Foundation under one or more agreements. | ||
| // The .NET Foundation licenses this file to you under the MIT license. | ||
|
|
||
| namespace Aspire.Hosting.ApplicationModel; | ||
|
|
||
| /// <summary> | ||
| /// A resource that represents a Kafka broker. | ||
| /// </summary> | ||
| /// <param name="name">The name of the resource.</param> | ||
| public class KafkaServerResource(string name) : Resource(name), IResourceWithConnectionString, IResourceWithEnvironment | ||
| { | ||
| /// <summary> | ||
| /// Gets the connection string for Kafka broker. | ||
| /// </summary> | ||
| /// <returns>A connection string for the Kafka in the form "host:port" to be passed as <see href="https://docs.confluent.io/platform/current/clients/confluent-kafka-dotnet/_site/api/Confluent.Kafka.ClientConfig.html#Confluent_Kafka_ClientConfig_BootstrapServers">BootstrapServers</see>.</returns> | ||
| public string? GetConnectionString() | ||
| { | ||
| if (!this.TryGetAllocatedEndPoints(out var allocatedEndpoints)) | ||
| { | ||
| throw new DistributedApplicationException($"Kafka resource \"{Name}\" does not have endpoint annotation."); | ||
| } | ||
|
|
||
| return allocatedEndpoints.SingleOrDefault()?.EndPointString; | ||
| } | ||
|
|
||
| internal int GetPort() | ||
| { | ||
| if (!this.TryGetAllocatedEndPoints(out var allocatedEndpoints)) | ||
| { | ||
| throw new DistributedApplicationException($"Kafka resource \"{Name}\" does not have endpoint annotation."); | ||
| } | ||
|
|
||
| return allocatedEndpoints.Single().Port; | ||
| } | ||
| } |
23 changes: 23 additions & 0 deletions
23
src/Components/Aspire.Confluent.Kafka/Aspire.Confluent.Kafka.csproj
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <TargetFramework>$(NetCurrent)</TargetFramework> | ||
| <IsPackable>true</IsPackable> | ||
| <PackageTags>$(ComponentCommonPackageTags) kafka</PackageTags> | ||
| <Description>Confluent.Kafka based Kafka generic consumer and producer that integrates with Aspire, including healthchecks and metrics.</Description> | ||
| <NoWarn>$(NoWarn);SYSLIB1100</NoWarn> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <Compile Include="..\Common\HealthChecksExtensions.cs" Link="HealthChecksExtensions.cs" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="AspNetCore.HealthChecks.Kafka" /> | ||
| <PackageReference Include="Confluent.Kafka" /> | ||
| <PackageReference Include="Microsoft.Extensions.Configuration.Binder" /> | ||
| <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" /> | ||
| <PackageReference Include="OpenTelemetry.Extensions.Hosting" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
Oops, something went wrong.