Hi, I was trying to subscribe to data change events for LEAF data services using the new C# library, at the moment I am trying to subscribe to Kafka topics to do this.
My data services are created using the standard yo generators, in my data service how do I know what Kafka topics the data change events are raised against? I am assuming it would be something like “.datachangeevent.add”
When we do it via LEAF JS work we can directly subscribe to web sockets so we do not need any special topic names etc.
Hello there! The default topic names for a given service are [servicename].[datachangeevents].[datatype]. For example, if your service prefix (can find this in your values.yml) is PersonService, the topic used would be:
personservice.datachangeevents.dataobject for DataObjects
personservice.datachangeevents.beanmetadata for BeanMetadata
personservice.datachangeevents.metadatatemplate for MetadataTemplates
personservice.datachangeevents.lookuplist for LookupLists
etc.
Note that topics are by default all lower case.
Also you can customize topics in Java Data Services and Functional Services using variables in the kafka section of your values.yml. For example to customize the DataObject and BeanMetadata topics, you would have something like:
kafka:
producer:
bootstrapServers: kafka.kafka:9092
consumer:
bootstrapServers: kafka.kafka:9092
dataObjectTopic: my.dataobject.topic
beanMetadataTopic: my.beanmetadata.topic
For the C# Functional Service, though, you’ll need to subscribe to the topics using the Kafka Consumer object.
1 Like
Thanks Mark, much appreciated. is it right to assume all actions against DataObjects like ADD, UPDATE and DELETE come through same topic “personservice.datachangeevents.dataobject” in this case?
Thanks
Yep that is correct! The default is ADD, UPDATE, and DELETE all go to the same topic.
That said, in your Java Data Services, you can supply a custom topicKeyStrategy, and actually route ADD events to one topic, UPDATE to another, etc. if needed. Here is a simple example below that does what the default topic strategy would do, but appends .add, .update, or .delete based on the event’s DataChangeType:
IDataChangeEventKafkaTopicKeyStrategy defaultTopicKeyStrategy = EventServiceKafkaTopicKeyStrategy.newBuilder("PersonService")
.build();
IDataChangeEventKafkaTopicKeyStrategy topicKeyStrategy = event -> {
Pair<String, Optional<String>> defaultTopicKey = defaultTopicKeyStrategy.getTopicAndKey(event);
String topic = defaultTopicKey.getLeft();
Optional<String> keyOpt = defaultTopicKey.getRight();
switch (event.getDataChangeType()) {
case ADD:
topic += ".add";
break;
case UPDATE:
topic += ".update";
break;
case DELETE:
topic += ".delete";
break;
}
return Pair.of(topic, keyOpt);
};
IEventService eventService = EventServiceKafka.newBuilder()
.withTopicKeyStrategy(topicKeyStrategy)
.build();
1 Like
