Flushless producer supporting both comparable and non comparable offsets #873

shrinandthakkar · 2021-11-30T07:12:37Z

In Mirrormaking, the connector job polls a list of records from the source and directs those records to be sent out to the destination via the transport provider using the flushless event producer. The flushless event producer keeps track of the in-flight messages and the acknowledged checkpoints for each source partition. These checkpoints are then used by the connector to commit the safe offsets to the source via the consumer API. And this is how the whole mirror maker process works in brief.

Before this change, this flushless event producer supported any pub-sub framework that has comparable type offsets. The offsets were compared by using a priority queue and based on that the checkpoints were moved in the flushless event producer's handler.

This new change provides support for both comparable offsets by using a priority queue and comparing offsets to move the checkpoint, and noncomparable offsets by using a simple deque-based approach that removes the dependency on comparable offsets for any underlying pub-sub framework.

Example Scenario:

Let's suppose that the connector reads messages from a source topic T, partition P from offsets L1, L2, L3 (in order) and the flushless event producer sends those messages out to the destination.

We receive a callback for the events in the order of L2, L1, L3 confirming that it is produced successfully in that order.

1. Flushless event producer with Comparable offsets

Callback for L2
{ L1, L2, L3 }
Inflights: L1, L3
Acked: L2

Since there is no acked checkpoint that is smaller than the oldest inflight checkpoint, there is no safe checkpoint to send back to the connector here.

Safe checkpoint: <none>
Callback for L1
{L1, L2, L3}
Inflights: L3
Acked: L1, L2

Following the algorithm the largest acked checkpoint that is still smaller than the oldest inflight checkpoint is L2. The FlushlessEventProducerHandler removes all checkpoints in acked list that are smaller than the oldest inflight checkpoint, and then returns the largest (L2) to the connector as the safe checkpoint for topic T partition P.

Safe checkpoint: L2
Callback for L3
{L3}
Inflights: {}
Acked: L3

Safe checkpoint: L3

2. Flushless event producer with Non Comparable offsets

Callback for L2
{ L1, L2, L3 }

Safe checkpoint: <none>
Callback for L1
{L1, L2, L3}

Since there is a contiguous prefix of acknowledged checkpoints, update the safe checkpoint to be the last removed checkpoint from the deque.

Safe checkpoint: L2
Callback for L3
{L3}

Since there is a contiguous prefix of acknowledged checkpoints, update the safe checkpoint to be the last removed checkpoint from the deque.

Safe checkpoint: L3

…ets for Ack-ing

...m-server-api/src/main/java/com/linkedin/datastream/server/FlushlessEventProducerHandler.java

...nector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaBasedConnectorConfig.java

...m-server-api/src/main/java/com/linkedin/datastream/server/FlushlessEventProducerHandler.java

vmaheshw · 2021-12-14T23:33:44Z

...rver-api/src/test/java/com/linkedin/datastream/server/TestFlushlessEventProducerHandler.java

@@ -32,10 +32,38 @@
  private static final String TOPIC = "MyTopic";
  private static final Random RANDOM = new Random();

+  private static final String OFFSET_CHECKPOINT_TRACKING_STRATEGY_WITH_COMPARABLE_OFFSETS =


Instead of redefining, you can reference these configs.

...api/src/main/java/com/linkedin/datastream/server/CallbackStatusWithNonComparableOffsets.java

...rver-api/src/test/java/com/linkedin/datastream/server/TestFlushlessEventProducerHandler.java

vmaheshw · 2021-12-20T22:00:53Z

...nector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaBasedConnectorConfig.java

@@ -110,6 +114,9 @@ public KafkaBasedConnectorConfig(Properties properties) {
        INCLUDE_DATASTREAM_NAME_IN_CONSUMER_CLIENT_ID, DEFAULT_INCLUDE_DATASTREAM_NAME_IN_CONSUMER_CLIENT_ID);
    _enablePartitionAssignment = verifiableProperties.getBoolean(ENABLE_PARTITION_ASSIGNMENT, Boolean.FALSE);

+    _callbackStatusStrategy = verifiableProperties.getString(CONFIG_CALLBACK_STATUS_STRATEGY_FACTORY_CLASS,
+        CallbackStatusWithComparableOffsetsFactory.class.getName());
+


You can add verification as well here and save the object rather then saving the name and putting the onus on the caller to verify and throw exception.

...nector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaBasedConnectorConfig.java

...java/com/linkedin/datastream/connectors/kafka/mirrormaker/KafkaMirrorMakerConnectorTask.java

...m-server-api/src/main/java/com/linkedin/datastream/server/FlushlessEventProducerHandler.java

...va/com/linkedin/datastream/server/callbackstatus/CallbackStatusWithNonComparableOffsets.java

.../java/com/linkedin/datastream/server/callbackstatus/CallbackStatusWithComparableOffsets.java

vmaheshw · 2021-12-20T22:18:42Z

...va/com/linkedin/datastream/server/callbackstatus/CallbackStatusWithNonComparableOffsets.java

+  // Hashset storing all the records which are yet to be acked
+  private final Set<T> _inFlight = Collections.synchronizedSet(new LinkedHashSet<>());
+
+  // Deque to store all the messages which are inflight until the last consumer checkpoint is made
+  private final Deque<T> _inFlightUntilLastConsumerCheckpoint = new ArrayDeque<>();


It is not clear how are both of these different. Can you please explain the logic?

This list maintains the records which are inflight until the latest checkpoint is made at the consumer side,

Example :
If we have 3 records for a topic partition that are consumed as L1, L2, L3, and we receive ack for L2, the _inFlights set would have 2 records, since we already got produce ack for L2, but the _inFlightUntilLastConsumerCheckpoint would have all 3 records as we still don't have a new consumer checkpoint to make until everything before L2 is acked from producer side.

...va/com/linkedin/datastream/server/callbackstatus/CallbackStatusWithNonComparableOffsets.java

vmaheshw · 2021-12-20T22:27:40Z

...va/com/linkedin/datastream/server/callbackstatus/CallbackStatusWithNonComparableOffsets.java

+      _currentCheckpoint = _inFlightUntilLastConsumerCheckpoint.pollFirst();
+
+      if (!_acked.remove(_currentCheckpoint)) {
+        LOG.error("Internal state error; could not remove checkpoint {}", _currentCheckpoint);


Just curious: How will currentCheckpoints will look in case of non-comparable offsets? How will we debug the issues, i.e. stuck partitions? Offset numbers were easy to compare.

The checkpoints in non-comparable scenarios could be of byte type.

...rver-api/src/test/java/com/linkedin/datastream/server/TestFlushlessEventProducerHandler.java

jzakaryan

lgtm

vmaheshw · 2022-01-05T19:28:42Z

...linkedin/datastream/server/callbackstatus/CallbackStatusWithNonComparableOffsetsFactory.java

@@ -6,11 +6,16 @@
 package com.linkedin.datastream.server.callbackstatus;

 /**
- * Interface for CallbackStatus Factories
+ * Factory implementation for Callback Status With Non Comparable Offsets


nit: lower case for Callback Status With Non Comparable Offsets

…ets (linkedin#873) * Flushless producer without comparing offsets * Flushless Producer Supporting Both Comparable and Non Comparable Offsets for Ack-ing * Fix comments * Making methods abstract * Created config support to use both strategies and added tests * Address comments * Address comments Co-authored-by: Shrinand Thakkar <sthakkar@sthakkar-mn1.linkedin.biz>

Shrinand Thakkar added 3 commits November 29, 2021 15:38

Flushless producer without comparing offsets

3a468e6

Flushless Producer Supporting Both Comparable and Non Comparable Offs…

bf64205

…ets for Ack-ing

Fix comments

33eebc2

shrinandthakkar requested review from vmaheshw and jzakaryan November 30, 2021 19:07

jzakaryan requested changes Dec 8, 2021

View reviewed changes

...m-server-api/src/main/java/com/linkedin/datastream/server/FlushlessEventProducerHandler.java Outdated Show resolved Hide resolved

Shrinand Thakkar added 2 commits December 9, 2021 15:03

Making methods abstract

1644f0c

Created config support to use both strategies and added tests

062ba47

jzakaryan previously approved these changes Dec 13, 2021

View reviewed changes

...nector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaBasedConnectorConfig.java Outdated Show resolved Hide resolved

vmaheshw reviewed Dec 14, 2021

View reviewed changes

Address comments

1ae2525

shrinandthakkar dismissed jzakaryan’s stale review via 1ae2525 December 17, 2021 07:15

vmaheshw reviewed Dec 20, 2021

View reviewed changes

...rver-api/src/test/java/com/linkedin/datastream/server/TestFlushlessEventProducerHandler.java Show resolved Hide resolved

...rver-api/src/test/java/com/linkedin/datastream/server/TestFlushlessEventProducerHandler.java Outdated Show resolved Hide resolved

Address comments

ade37ff

jzakaryan approved these changes Dec 22, 2021

View reviewed changes

vmaheshw self-requested a review January 5, 2022 19:20

vmaheshw approved these changes Jan 5, 2022

View reviewed changes

jzakaryan merged commit 039c28c into linkedin:master Jan 5, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Flushless producer supporting both comparable and non comparable offsets #873

Flushless producer supporting both comparable and non comparable offsets #873

shrinandthakkar commented Nov 30, 2021 •

edited

Loading

vmaheshw Dec 14, 2021

vmaheshw Dec 20, 2021

vmaheshw Dec 20, 2021

shrinandthakkar Dec 21, 2021

vmaheshw Dec 20, 2021

shrinandthakkar Dec 22, 2021

jzakaryan left a comment

vmaheshw Jan 5, 2022

Flushless producer supporting both comparable and non comparable offsets #873

Flushless producer supporting both comparable and non comparable offsets #873

Conversation

shrinandthakkar commented Nov 30, 2021 • edited Loading

Example Scenario:

1. Flushless event producer with Comparable offsets

2. Flushless event producer with Non Comparable offsets

vmaheshw Dec 14, 2021

Choose a reason for hiding this comment

vmaheshw Dec 20, 2021

Choose a reason for hiding this comment

vmaheshw Dec 20, 2021

Choose a reason for hiding this comment

shrinandthakkar Dec 21, 2021

Choose a reason for hiding this comment

vmaheshw Dec 20, 2021

Choose a reason for hiding this comment

shrinandthakkar Dec 22, 2021

Choose a reason for hiding this comment

jzakaryan left a comment

Choose a reason for hiding this comment

vmaheshw Jan 5, 2022

Choose a reason for hiding this comment

shrinandthakkar commented Nov 30, 2021 •

edited

Loading