pabloem commented on code in PR #24713:
URL: https://github.com/apache/beam/pull/24713#discussion_r1109090752
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:
##########
@@ -202,6 +209,63 @@ private static Instant ensureTimestampWithinBounds(Instant
timestamp) {
return timestamp;
}
+ private static final Cache<Map<String, String>, SourceTask> CONNECTOR_CACHE =
+ CacheBuilder.newBuilder()
+ .expireAfterAccess(java.time.Duration.ofSeconds(60))
+ .removalListener(
+ new RemovalListener<Map<String, String>, SourceTask>() {
+ @Override
+ public void onRemoval(
+ RemovalNotification<Map<String, String>, SourceTask>
removalNotification) {
+ LOG.debug(
+ "Task for key [[{}]] is being removed. Cause: {}",
+ removalNotification.getKey(),
+ removalNotification.getCause());
+ removalNotification.getValue().stop();
+ }
+ })
+ .maximumSize(10)
+ .build();
+
+ SourceTask getDebeziumSourceTask(
+ Map<String, String> element, RestrictionTracker<OffsetHolder,
Map<String, Object>> tracker) {
+ final KafkaSourceConsumerFn<T> dofnInstance = this;
+ try {
Review Comment:
we don't need to. The task should always have an up-to-date offset with the
latest offset. However I want to test it in the next PR (add some sort of
asserttion on the latest initialized context)
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:
##########
@@ -202,6 +209,63 @@ private static Instant ensureTimestampWithinBounds(Instant
timestamp) {
return timestamp;
}
+ private static final Cache<Map<String, String>, SourceTask> CONNECTOR_CACHE =
+ CacheBuilder.newBuilder()
+ .expireAfterAccess(java.time.Duration.ofSeconds(60))
+ .removalListener(
+ new RemovalListener<Map<String, String>, SourceTask>() {
+ @Override
+ public void onRemoval(
+ RemovalNotification<Map<String, String>, SourceTask>
removalNotification) {
+ LOG.debug(
+ "Task for key [[{}]] is being removed. Cause: {}",
+ removalNotification.getKey(),
+ removalNotification.getCause());
+ removalNotification.getValue().stop();
+ }
+ })
+ .maximumSize(10)
+ .build();
+
+ SourceTask getDebeziumSourceTask(
+ Map<String, String> element, RestrictionTracker<OffsetHolder,
Map<String, Object>> tracker) {
Review Comment:
okay I tried to do this wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
