1+ import {
2+ AvroSerializer , AvroSerializerConfig , SerdeType ,
3+ ClientConfig , SchemaRegistryClient , SchemaInfo , BearerAuthCredentials
4+ } from "@confluentinc/schemaregistry" ;
5+ import { CreateAxiosDefaults } from "axios" ;
6+ import { KafkaJS } from '@confluentinc/kafka-javascript' ;
7+ import {
8+ clusterBootstrapUrl ,
9+ baseUrl ,
10+ issuerEndpointUrl , oauthClientId , oauthClientSecret , scope ,
11+ identityPoolId , schemaRegistryLogicalCluster , kafkaLogicalCluster
12+ } from "./constants" ;
13+ import axios from 'axios' ;
14+
15+ // Only showing the producer, will be the same implementation for the consumer
16+
17+ async function token_refresh ( ) {
18+ try {
19+ // Make a POST request to get the access token
20+ const response = await axios . post ( issuerEndpointUrl , new URLSearchParams ( {
21+ grant_type : 'client_credentials' ,
22+ client_id : oauthClientId ,
23+ client_secret : oauthClientSecret ,
24+ scope : scope
25+ } ) , {
26+ headers : {
27+ 'Content-Type' : 'application/x-www-form-urlencoded'
28+ }
29+ } ) ;
30+
31+ // Extract the token and expiration time from the response
32+ const token = response . data . access_token ;
33+ const exp_seconds = Math . floor ( Date . now ( ) / 1000 ) + response . data . expires_in ;
34+ const exp_ms = exp_seconds * 1000 ;
35+
36+ const principal = 'admin' ; // You can adjust this based on your needs
37+ const extensions = {
38+ traceId : '123' ,
39+ logicalCluster : kafkaLogicalCluster ,
40+ identityPoolId : identityPoolId
41+ } ;
42+
43+ return { value : token , lifetime : exp_ms , principal, extensions } ;
44+ } catch ( error ) {
45+ console . error ( 'Failed to retrieve OAuth token:' , error ) ;
46+ throw new Error ( 'Failed to retrieve OAuth token' ) ;
47+ }
48+ }
49+
50+ async function kafkaProducerAvro ( ) {
51+
52+ const createAxiosDefaults : CreateAxiosDefaults = {
53+ timeout : 10000
54+ } ;
55+
56+ const bearerAuthCredentials : BearerAuthCredentials = {
57+ credentialsSource : 'OAUTHBEARER' ,
58+ issuerEndpointUrl : issuerEndpointUrl ,
59+ clientId : oauthClientId ,
60+ clientSecret : oauthClientSecret ,
61+ scope : scope ,
62+ identityPoolId : identityPoolId ,
63+ logicalCluster : schemaRegistryLogicalCluster
64+ }
65+
66+ const clientConfig : ClientConfig = {
67+ baseURLs : [ baseUrl ] ,
68+ createAxiosDefaults : createAxiosDefaults ,
69+ cacheCapacity : 512 ,
70+ cacheLatestTtlSecs : 60 ,
71+ bearerAuthCredentials : bearerAuthCredentials
72+ } ;
73+
74+ const schemaRegistryClient = new SchemaRegistryClient ( clientConfig ) ;
75+
76+ const kafka : KafkaJS . Kafka = new KafkaJS . Kafka ( {
77+ kafkaJS : {
78+ brokers : [ clusterBootstrapUrl ] ,
79+ ssl : true ,
80+ sasl : {
81+ mechanism : 'oauthbearer' ,
82+ oauthBearerProvider : token_refresh
83+ } ,
84+ } ,
85+ } ) ;
86+
87+ const producer : KafkaJS . Producer = kafka . producer ( {
88+ kafkaJS : {
89+ allowAutoTopicCreation : true ,
90+ acks : 1 ,
91+ compression : KafkaJS . CompressionTypes . GZIP ,
92+ }
93+ } ) ;
94+
95+ console . log ( "Producer created" ) ;
96+
97+ const schemaString : string = JSON . stringify ( {
98+ type : 'record' ,
99+ name : 'User' ,
100+ fields : [
101+ { name : 'name' , type : 'string' } ,
102+ { name : 'age' , type : 'int' } ,
103+ ] ,
104+ } ) ;
105+
106+ const schemaInfo : SchemaInfo = {
107+ schemaType : 'AVRO' ,
108+ schema : schemaString ,
109+ } ;
110+
111+ const userTopic = 'example-user-topic' ;
112+ await schemaRegistryClient . register ( userTopic + "-value" , schemaInfo ) ;
113+
114+ const userInfo = { name : 'Alice N Bob' , age : 30 } ;
115+
116+ const avroSerializerConfig : AvroSerializerConfig = { useLatestVersion : true } ;
117+
118+ const serializer : AvroSerializer = new AvroSerializer ( schemaRegistryClient , SerdeType . VALUE , avroSerializerConfig ) ;
119+
120+ const outgoingMessage = {
121+ key : "1" ,
122+ value : await serializer . serialize ( userTopic , userInfo )
123+ } ;
124+
125+ console . log ( "Outgoing message: " , outgoingMessage ) ;
126+
127+ await producer . connect ( ) ;
128+
129+ await producer . send ( {
130+ topic : userTopic ,
131+ messages : [ outgoingMessage ]
132+ } ) ;
133+
134+ await producer . disconnect ( ) ;
135+ }
136+
137+ kafkaProducerAvro ( ) ;
0 commit comments