Kafka connector, dynamic datetime in body voor REST call

[Douweegbertje]

Ik gebruik: https://github.com/llofberg/kafka-connect-rest voor een relatief simpele REST call. Nu wil ik eigenlijk alleen recent geüpdatet data binnen halen. Ik kan middels een body JSON mee sturen met parameters. O.a. ook voor de 'updated time' veld.

Dus bijvoorbeeld zo:

code:

rest.source.body={"updated time":"24-07-19 08:00"}

Het gaat hier trouwens om de .properties van de connector

Nu vraag ik mij eigenlijk af hoe ik er voor kan zorgen dat de value van de datetime gewoon "nu minus 2 minuten" is, in dat geval kan ik om de minuut pollen en exact die data ophalen die gewijzigd was.

Ik heb diverse dingen gezocht maar dan kom je met name uit hoe je "variables" krijgt t.b.v. converten. Bijv. in je sink en/of messages die je naar kafka wilt sturen. Mij gaat het dus eigenlijk andersom; hoe krijg ik een dynamische waarde in een actie die mijn connector doet.

[YakuzA]

Bij veel kafka libraries heb je een kafka consumer die via een autocommit bijhoud waar je bent in de stream, in een consume die hij opvraagd bij de kafka broker krijg je dan alleen de mutaties sinds je laatste connect terug. (tenzij je zelf iets forceert dat je weer een complete sync wil)

(weet niet hoe dat zit in de specifieke connector die je gebruikt. Gebruiken hier de confluent.kafka )

[ Voor 14% gewijzigd door YakuzA op 24-07-2019 12:17 ]

[Douweegbertje]

YakuzA schreef op woensdag 24 juli 2019 @ 11:47:
Bij veel kafka libraries heb je een kafka consumer die via een autocommit bijhoud waar je bent in de stream, in een consume die hij opvraagd bij de kafka broker krijg je dan alleen de mutaties sinds je laatste connect terug. (tenzij je zelf iets forceert dat je weer een complete sync wil)

(weet niet hoe dat zit in de specifieke connector die je gebruikt. Gebruiken hier de confluent.kafka )

Alleen dit is aan de producer kant, die zelf data ophaalt bij een source. Ik wil de "arguments" die hij gebruikt voor het ophalen dynamisch kunnen bepalen d.m.v. in dit geval dus een datetime veld.

[YakuzA]

Douweegbertje schreef op woensdag 24 juli 2019 @ 12:39:
[...]
Alleen dit is aan de producer kant, die zelf data ophaalt bij een source. Ik wil de "arguments" die hij gebruikt voor het ophalen dynamisch kunnen bepalen d.m.v. in dit geval dus een datetime veld.

Je hebt dus een Producer met daarin geïntegreerd een Consumer?
Ik begrijp hem nog niet helemaal

Een Producer zet namelijk alleen zelf iets op kafka 'bus' richting de broker en zal niks te doen hebben met wat er reeds op kafka aanwezig is.
Een Consumer haalt juist informatie van kafka af.

[Douweegbertje]

YakuzA schreef op woensdag 24 juli 2019 @ 13:17:
[...]

Je hebt dus een Producer met daarin geïntegreerd een Consumer?
Ik begrijp hem nog niet helemaal

Een Producer zet namelijk alleen zelf iets op kafka 'bus' richting de broker en zal niks te doen hebben met wat er reeds op kafka aanwezig is.
Een Consumer haalt juist informatie van kafka af.

Ik weet niet waarom je het stukje consumer erbij haalt, dat staat nergens

Het gaat om een connector die dus als producer functioneert en data ophaalt vanuit een source (een website bijv.) en dit in Kafka zet. Ik wil in de configuratie van de connector iets kunnen opgeveven als "currenttime" als argument voor elke task die hij doet.

[YakuzA]

Douweegbertje schreef op woensdag 24 juli 2019 @ 13:46:
[...]


Ik weet niet waarom je het stukje consumer erbij haalt, dat staat nergens

Het gaat om een connector die dus als producer functioneert en data ophaalt vanuit een source (een website bijv.) en dit in Kafka zet. Ik wil in de configuratie van de connector iets kunnen opgeveven als "currenttime" als argument voor elke task die hij doet.

Je probleem heeft dus niks met kafka te maken?

Maar gewoon DateTime.Now.AddMinutes(-2).ToString() aan een parameter knopen en je bent klaar?