kafka produce #8664
kafka produce #8664
Conversation
| @@ -396,6 +397,7 @@ export async function batchWriteWithAlerts( | |||
| const filteredItems: AWS.DynamoDB.DocumentClient.PutItemInputAttributeMap[] = | |||
| await checkMovement(items, previousItems); | |||
| await batchWrite(filteredItems, failOnError); | |||
| await produceKafkaTopics(filteredItems.filter((i) => i.SK == 0) as any[]); | |||
There was a problem hiding this comment.
this would mean, we write only current records? also, topic is missing here. Guessing this is sample code?
So, other than minor comments, the code looks good, we should give it a try and write some to kafka, so we can test writing to our DB
There was a problem hiding this comment.
the batch writes to dynamo are all duplicated (sk 0 and historical). Here I've filtered the items to sk0 only and then written them to all topics. This will write all topics.
There was a problem hiding this comment.
Hmm, I would rather split between 0 & with timestamp, write the timestamp ones too time series & without ts to current, I guess the mapping for both needs to be sent to mapping topic?
There was a problem hiding this comment.
My worry is, what happens when you refill coin data, then it won't be sent to the topics right? Like filling nft prices a month ago
coins/src/utils/coins3/produce.ts
Outdated
| @@ -23,7 +23,7 @@ async function produceTopics( | |||
|
|
|||
| items.map((item) => { | |||
| const { symbol, decimals } = item; | |||
| if (!symbol || !decimals) return; | |||
| if (!symbol || decimals == null) return; | |||
There was a problem hiding this comment.
typeof decimals !== 'number' is safer? or it will skip decimals which are string? hmm... maybe we can leave this as is
No description provided.