[CDAP-21079] Implement cdap-tms-spanner extension - Part 1 #15750
Conversation
| @@ -139,13 +142,25 @@ private MessagingService getDelegate() { | |||
| LOG.info("Messaging service {} is loaded", messagingService.getName()); | |||
| try { | |||
| messagingService.initialize(new DefaultMessagingServiceContext(this.cConf)); | |||
| } catch (IOException e) { | |||
| createTopics(cConf, messagingService); | |||
There was a problem hiding this comment.
I assume we need this for Spanner implementation. It's a bit strange to create topic here. How are we creating topics for TMS implementation now?
There was a problem hiding this comment.
Why not putting this in initialize() method inside spanner implementation?
There was a problem hiding this comment.
For TMS implementation, it is part of the start up which is called when MessagingService pod comes up : CoreMessagingService.java#L239
I was planning to do createTopics() in SpannerMessagingService.initialise() but the issue here is the MessagingServiceUtils class is also in cdap-tms. Also, DefaultTopicMetadata which is input to the createTopic() method. This class is in cdap-tms module (same level as the DefaultMessagingServiceContext, DelegatingMessagingService).
We can somehow get away with the DefaultTopicMetadata dependency, as the properties which are set here are not very useful in case of Cloud Spanner. But the MessagingServiceUtils is needed here.
I have added some code comments in the createTopics(), may be, I can add a TODO as well there stating this method can be removed from DelegatingMessagingService after the cdap-tms module refactoring.
cdap-tms/src/main/java/io/cdap/cdap/messaging/service/CoreMessagingService.java
Outdated
Show resolved
Hide resolved
...saging-ext-spanner/src/main/java/io/cdap/cdap/messaging/spanner/SpannerMessagingService.java
Outdated
Show resolved
Hide resolved
...saging-ext-spanner/src/main/java/io/cdap/cdap/messaging/spanner/SpannerMessagingService.java
Outdated
Show resolved
Hide resolved
...saging-ext-spanner/src/main/java/io/cdap/cdap/messaging/spanner/SpannerMessagingService.java
Outdated
Show resolved
Hide resolved
...saging-ext-spanner/src/main/java/io/cdap/cdap/messaging/spanner/SpannerMessagingService.java
Outdated
Show resolved
Hide resolved
...saging-ext-spanner/src/main/java/io/cdap/cdap/messaging/spanner/SpannerMessagingService.java
Outdated
Show resolved
Hide resolved
...saging-ext-spanner/src/main/java/io/cdap/cdap/messaging/spanner/SpannerMessagingService.java
Outdated
Show resolved
Hide resolved
|
Implement methods like createTopic(), publish() etc.
Follow up PR to include changes for fetch().
createAllTopics() is placed in DelegatingMessagingService() right now as it has dependency on DefaultTopicMetadata class that comes from cdap-tms module. Extensions should not have this dependency. This is a TODO where cdap-tms would be refactored to remove these dependencies.
The code is tested in autopush project.
Junits are tested by setting the system properties, as done in SpannerStructuredAdminTest.
