Implement cdap-tms-spanner extension - Part 1 #15730
Conversation
|
|
||
| @Override | ||
| public Map<String, String> getProperties() { | ||
| String spannerPropertiesPrefix = |
There was a problem hiding this comment.
TMS is one implementation of the messaging SPI. The other implementation will be spanner. So TMS should not have any reference to spanner.
| * | ||
| * @param messagingContext messaging service context | ||
| */ | ||
| void destroy(MessagingContext messagingContext); |
There was a problem hiding this comment.
Why this? When do you want to close spanner? We probably should keep the client open in spanner messaging service in appfabric, otherwise it'll hurt the performance substantially.
| + " (OLDER_THAN(publish_ts, INTERVAL 7 DAY))", getTableName(topicId), SEQUENCE_ID_FIELD, | ||
| PAYLOAD_SEQUENCE_ID, PUBLISH_TS_FIELD, PAYLOAD_FIELD); | ||
| OperationFuture<Void, UpdateDatabaseDdlMetadata> future = adminClient.updateDatabaseDdl( | ||
| SpannerUtil.getInstanceID(cConf), SpannerUtil.getDatabaseID(cConf), |
There was a problem hiding this comment.
initialize instanceId, databaseId, etc at the beginning and reuse those fields. It'll be much simpler than calling SpannerUtil.getInstanceID(cConf), SpannerUtil.getDatabaseID(cConf), etc
| try { | ||
| future.get(); | ||
| } catch (InterruptedException | ExecutionException e) { | ||
| LOG.error("Error when executing {}", topicSQL, e); |
There was a problem hiding this comment.
We should not log and forget the exception. Check the handleError in ClientMessagingService
| } | ||
| } | ||
|
|
||
| public static String getTableName(TopicId topicId) { |
| try { | ||
| Thread.sleep(5); | ||
| } catch (InterruptedException e) { | ||
| LOG.error("error during sleep", e); |
There was a problem hiding this comment.
Do not log and forget exceptions
|
This PR contains initial implementation of spanner extension of messaging service.
For now conf property : messaging.service.name will decide if spanner messaging service will be used. This part resides in the DelegatingMessagingService implemented earlier.
Follow up PR to contain implementation of other methods in the SpannerMessagingService : especially fetch() and appropriately call destroy() to close spanner clients.
Reference Masoud's POC : develop...features/spanner-tms