You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2023/01/18 23:16:49 UTC
[GitHub] [beam] lukecwik commented on a diff in pull request #24713: Caching DebeziumIO SourceTask instances to reduce the number of outstanding connections
lukecwik commented on code in PR #24713:
URL: https://github.com/apache/beam/pull/24713#discussion_r1080655146
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:
##########
@@ -217,26 +239,44 @@ private static Instant ensureTimestampWithinBounds(Instant timestamp) {
public ProcessContinuation process(
@Element Map<String, String> element,
RestrictionTracker<OffsetHolder, Map<String, Object>> tracker,
- OutputReceiver<T> receiver)
- throws Exception {
- Map<String, String> configuration = new HashMap<>(element);
-
+ OutputReceiver<T> receiver) {
// Adding the current restriction to the class object to be found by the database history
register(tracker);
+ Map<String, String> configuration = new HashMap<>(element);
+ String cacheKey =
+ configuration.entrySet().stream()
Review Comment:
nit: note that if this is a hot path then `.stream()` is 2x slower then a for loop in many use cases.
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:
##########
@@ -217,26 +239,44 @@ private static Instant ensureTimestampWithinBounds(Instant timestamp) {
public ProcessContinuation process(
@Element Map<String, String> element,
RestrictionTracker<OffsetHolder, Map<String, Object>> tracker,
- OutputReceiver<T> receiver)
- throws Exception {
- Map<String, String> configuration = new HashMap<>(element);
-
+ OutputReceiver<T> receiver) {
// Adding the current restriction to the class object to be found by the database history
register(tracker);
+ Map<String, String> configuration = new HashMap<>(element);
+ String cacheKey =
+ configuration.entrySet().stream()
+ .map(entry -> String.format("(%s,%s)", entry.getKey(), entry.getValue()))
+ .sorted()
+ .collect(Collectors.joining(","));
configuration.put(BEAM_INSTANCE_PROPERTY, this.getHashCode());
+ SourceTask task;
- SourceConnector connector = connectorClass.getDeclaredConstructor().newInstance();
- connector.start(configuration);
-
- SourceTask task = (SourceTask) connector.taskClass().getDeclaredConstructor().newInstance();
-
- try {
+ if (CONNECTOR_CACHE.getIfPresent(cacheKey) == null) {
Review Comment:
Consider using https://guava.dev/releases/26.0-jre/api/docs/com/google/common/cache/Cache.html#get-K-java.util.concurrent.Callable- and having the callable be responsible for creating a new connector
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:
##########
@@ -217,26 +239,44 @@ private static Instant ensureTimestampWithinBounds(Instant timestamp) {
public ProcessContinuation process(
@Element Map<String, String> element,
RestrictionTracker<OffsetHolder, Map<String, Object>> tracker,
- OutputReceiver<T> receiver)
- throws Exception {
- Map<String, String> configuration = new HashMap<>(element);
-
+ OutputReceiver<T> receiver) {
// Adding the current restriction to the class object to be found by the database history
register(tracker);
+ Map<String, String> configuration = new HashMap<>(element);
+ String cacheKey =
+ configuration.entrySet().stream()
+ .map(entry -> String.format("(%s,%s)", entry.getKey(), entry.getValue()))
Review Comment:
Why convert this to a string, can you use `element` as the key?
Note that java map equality is defined as
```
Compares the specified object with this map for equality. Returns true if the given object is also a map and the two maps represent the same mappings. More formally, two maps m1 and m2 represent the same mappings if m1.entrySet().equals(m2.entrySet()). This ensures that the equals method works properly across different implementations of the Map interface.
```
Note that elements are immutable and are typically safe to be store beyond the lifetime of a bundle.
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:
##########
@@ -217,26 +239,44 @@ private static Instant ensureTimestampWithinBounds(Instant timestamp) {
public ProcessContinuation process(
@Element Map<String, String> element,
RestrictionTracker<OffsetHolder, Map<String, Object>> tracker,
- OutputReceiver<T> receiver)
- throws Exception {
- Map<String, String> configuration = new HashMap<>(element);
-
+ OutputReceiver<T> receiver) {
// Adding the current restriction to the class object to be found by the database history
register(tracker);
+ Map<String, String> configuration = new HashMap<>(element);
Review Comment:
If the common case is that the configuration is only needed for creating a new instance move the new hashmap call to when we actually need to create a copy and instead rely on generating the cache key from `element`
##########
sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/KafkaSourceConsumerFnTest.java:
##########
@@ -103,6 +103,7 @@ public void testStoppableKafkaSourceConsumerFn() {
.setCoder(VarIntCoder.of());
pipeline.run().waitUntilFinish();
+ // Since we're now caching calls
Review Comment:
Checking for how many times this is called will be flaky since it will all depend on how hot the cache remains, is checking how many times this is called important or should we drop this?
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:
##########
@@ -217,26 +239,44 @@ private static Instant ensureTimestampWithinBounds(Instant timestamp) {
public ProcessContinuation process(
@Element Map<String, String> element,
RestrictionTracker<OffsetHolder, Map<String, Object>> tracker,
- OutputReceiver<T> receiver)
- throws Exception {
- Map<String, String> configuration = new HashMap<>(element);
-
+ OutputReceiver<T> receiver) {
// Adding the current restriction to the class object to be found by the database history
register(tracker);
+ Map<String, String> configuration = new HashMap<>(element);
+ String cacheKey =
+ configuration.entrySet().stream()
+ .map(entry -> String.format("(%s,%s)", entry.getKey(), entry.getValue()))
+ .sorted()
+ .collect(Collectors.joining(","));
configuration.put(BEAM_INSTANCE_PROPERTY, this.getHashCode());
+ SourceTask task;
- SourceConnector connector = connectorClass.getDeclaredConstructor().newInstance();
- connector.start(configuration);
-
- SourceTask task = (SourceTask) connector.taskClass().getDeclaredConstructor().newInstance();
-
- try {
+ if (CONNECTOR_CACHE.getIfPresent(cacheKey) == null) {
+ SourceConnector connector = null;
+ try {
+ connector = connectorClass.getDeclaredConstructor().newInstance();
+ connector.start(configuration);
+ task = (SourceTask) connector.taskClass().getDeclaredConstructor().newInstance();
+ } catch (InstantiationException
+ | IllegalAccessException
+ | InvocationTargetException
+ | NoSuchMethodException e) {
+ throw new RuntimeException("Unable to initialize connector instance for Debezium", e);
+ }
Map<String, ?> consumerOffset = tracker.currentRestriction().offset;
- LOG.debug("--------- Consumer offset from Debezium Tracker: {}", consumerOffset);
+ LOG.debug("--------- Created new Debezium task with offset: {}", consumerOffset);
task.initialize(new BeamSourceTaskContext(tracker.currentRestriction().offset));
task.start(connector.taskConfigs(1).get(0));
+ CONNECTOR_CACHE.put(cacheKey, task);
+ } else {
+ task = CONNECTOR_CACHE.getIfPresent(cacheKey);
Review Comment:
this may have been evicted from the cache since you first checked which is why using get with a callable is needed
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org