-
Notifications
You must be signed in to change notification settings - Fork 896
DGS-23025 Add AssociatedNameStrategy #2577
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
29 commits
Select commit
Hold shift + click to select a range
822da62
First cut at AssociatedNameStrategy refactor
rayokota 4bab228
Minor fix
rayokota 827a334
Minor fix
rayokota 9d22fec
Fix async usage
rayokota 27a602d
Implement associated name strategy
rayokota 54c081a
Minor refactor
rayokota a524eb6
Minor cleanup
rayokota 748016a
Add more assoc apis
rayokota 1cf3782
handle 404
rayokota 2a8141a
Minor opt
rayokota a8b5e8a
Minor renaming
rayokota 4d3d8a6
Incorporate review feedback
rayokota 96f1d99
Incorporate review feedback
rayokota 9e3dc5a
Add tests
rayokota 33a6e62
Fix RecordType handling in subject name strategy
rayokota 890a30b
Fix AssociatedNameStrategy caching
rayokota d59e325
Minor fix
rayokota 298c191
Minor fix
rayokota d8251be
Minor fix
rayokota 0b3eefe
Make Associated the default strategy
rayokota 851b1c3
Fallback if sr client is null
rayokota a698717
Minor fixes
rayokota 55db7ce
Set default in serdes
rayokota 737abfc
Move setting of strategy before config null check
rayokota e860739
Move setting of strategy before config null check
rayokota c36e97c
Update docs
rayokota 0e9d8a2
Minor renaming
rayokota f6e2485
Add association example
rayokota f350f9d
Minor cleanup
rayokota File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or 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
13 changes: 13 additions & 0 deletions
13
examples/AvroGenericAssociation/AvroGenericAssociation.csproj
This file contains hidden or 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,13 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <ProjectTypeGuids>{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}</ProjectTypeGuids> | ||
| <AssemblyName>AvroGenericAssociation</AssemblyName> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <!-- nuget package reference: <PackageReference Include="Confluent.SchemaRegistry.Serdes.Avro" Version="2.9.0" /> --> | ||
| <ProjectReference Include="../../src/Confluent.SchemaRegistry.Serdes.Avro/Confluent.SchemaRegistry.Serdes.Avro.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
This file contains hidden or 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,177 @@ | ||
| // Copyright 2025 Confluent Inc. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
| // | ||
| // Refer to LICENSE for more information. | ||
|
|
||
| using Avro; | ||
| using Avro.Generic; | ||
| using Confluent.Kafka.SyncOverAsync; | ||
| using Confluent.SchemaRegistry; | ||
| using Confluent.SchemaRegistry.Serdes; | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Schema = Confluent.SchemaRegistry.Schema; | ||
|
|
||
|
|
||
| namespace Confluent.Kafka.Examples.AvroGenericAssociation | ||
| { | ||
| /// <summary> | ||
| /// Demonstrates how to use the Associated subject name strategy with | ||
| /// Avro GenericRecord serialization. This example: | ||
| /// 1. Registers a schema under a custom subject name. | ||
| /// 2. Creates a STRONG association between the topic and the subject. | ||
| /// 3. Produces and consumes messages using the Associated strategy | ||
| /// (which looks up the subject via the association). | ||
| /// 4. Deletes the association. | ||
| /// </summary> | ||
| class Program | ||
| { | ||
| static async Task Main(string[] args) | ||
| { | ||
| if (args.Length != 3) | ||
| { | ||
| Console.WriteLine("Usage: .. bootstrapServers schemaRegistryUrl topicName"); | ||
| return; | ||
| } | ||
|
|
||
| string bootstrapServers = args[0]; | ||
| string schemaRegistryUrl = args[1]; | ||
| string topicName = args[2]; | ||
| string groupName = "avro-generic-association-example-group"; | ||
| string subjectName = $"{topicName}-custom-value"; | ||
|
|
||
| var s = (RecordSchema)RecordSchema.Parse( | ||
| @"{ | ||
| ""type"": ""record"", | ||
| ""name"": ""User"", | ||
| ""fields"": [ | ||
| {""name"": ""name"", ""type"": ""string""}, | ||
| {""name"": ""favorite_number"", ""type"": ""long""}, | ||
| {""name"": ""favorite_color"", ""type"": ""string""} | ||
| ] | ||
| }" | ||
| ); | ||
|
|
||
| using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { Url = schemaRegistryUrl })) | ||
| { | ||
| // Step 1: Register the schema under a custom subject name. | ||
| var schemaObj = new Schema(s.ToString(), new List<SchemaReference>(), SchemaType.Avro); | ||
| var registeredSchema = await schemaRegistry.RegisterSchemaWithResponseAsync(subjectName, schemaObj, false); | ||
| Console.WriteLine($"Registered schema under subject '{subjectName}' with id {registeredSchema.Id}"); | ||
|
|
||
| // Step 2: Create a STRONG association between the topic and the subject. | ||
| var associationRequest = new AssociationCreateOrUpdateRequest( | ||
| resourceName: topicName, | ||
| resourceNamespace: "lkc-123", | ||
| resourceId: "lkc-123:" + topicName, | ||
| resourceType: "topic", | ||
| associations: new List<AssociationCreateOrUpdateInfo> | ||
| { | ||
| new AssociationCreateOrUpdateInfo( | ||
| subject: subjectName, | ||
| associationType: "value", | ||
| lifecycle: "STRONG", | ||
| frozen: null, | ||
| schema: null, | ||
| normalize: null | ||
| ) | ||
| } | ||
| ); | ||
| var associationResponse = await schemaRegistry.CreateAssociationAsync(associationRequest); | ||
| Console.WriteLine($"Created STRONG association: topic '{topicName}' -> subject '{subjectName}'"); | ||
|
|
||
| // Step 3: Produce and consume using the Associated strategy (the default). | ||
| CancellationTokenSource cts = new CancellationTokenSource(); | ||
| var consumeTask = Task.Run(() => | ||
| { | ||
| using (var consumer = | ||
| new ConsumerBuilder<string, GenericRecord>(new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = groupName, AutoOffsetReset = AutoOffsetReset.Earliest }) | ||
| .SetValueDeserializer(new AvroDeserializer<GenericRecord>(schemaRegistry).AsSyncOverAsync()) | ||
| .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}")) | ||
| .Build()) | ||
| { | ||
| consumer.Subscribe(topicName); | ||
|
|
||
| try | ||
| { | ||
| while (true) | ||
| { | ||
| try | ||
| { | ||
| var consumeResult = consumer.Consume(cts.Token); | ||
| Console.WriteLine($"Consumed - Key: {consumeResult.Message.Key}, Value: {consumeResult.Message.Value}"); | ||
| } | ||
| catch (ConsumeException e) | ||
| { | ||
| Console.WriteLine($"Consume error: {e.Error.Reason}"); | ||
| } | ||
| } | ||
| } | ||
| catch (OperationCanceledException) | ||
| { | ||
| consumer.Close(); | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| using (var producer = | ||
| new ProducerBuilder<string, GenericRecord>(new ProducerConfig { BootstrapServers = bootstrapServers }) | ||
| .SetValueSerializer(new AvroSerializer<GenericRecord>(schemaRegistry, new AvroSerializerConfig | ||
| { | ||
| AutoRegisterSchemas = false, | ||
| UseLatestVersion = true | ||
| })) | ||
| .Build()) | ||
| { | ||
| Console.WriteLine($"{producer.Name} producing on {topicName}. Enter user names, q to exit."); | ||
|
|
||
| long i = 1; | ||
| string text; | ||
| while ((text = Console.ReadLine()) != "q") | ||
| { | ||
| var record = new GenericRecord(s); | ||
| record.Add("name", text); | ||
| record.Add("favorite_number", i++); | ||
| record.Add("favorite_color", "blue"); | ||
|
|
||
| try | ||
| { | ||
| var dr = await producer.ProduceAsync(topicName, new Message<string, GenericRecord> { Key = text, Value = record }); | ||
| Console.WriteLine($"Produced to: {dr.TopicPartitionOffset}"); | ||
| } | ||
| catch (ProduceException<string, GenericRecord> ex) | ||
| { | ||
| Console.WriteLine($"Error producing message: {ex}"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| cts.Cancel(); | ||
| await consumeTask; | ||
|
|
||
| // Step 4: Delete the association. | ||
| var associations = await schemaRegistry.GetAssociationsByResourceNameAsync( | ||
| topicName, "-", "topic", null, null, 0, -1); | ||
| if (associations.Count > 0) | ||
| { | ||
| await schemaRegistry.DeleteAssociationsAsync( | ||
| associations[0].ResourceId, "topic", new List<string> { "value" }, true); | ||
| Console.WriteLine($"Deleted associations for topic '{topicName}'"); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.