[Pulsar SQL] Add max split entry queue size bytes limitation #9628

gaoran10 · 2021-02-19T10:02:45Z

Motivation

In Pulsar SQL, there are two configurations pulsar.max-split-entry-queue-size and pulsar.max-split-message-queue-size to control the entry queue and message queue capacity, but some entries are so big some are small, it's hard to control the queue size bytes and the message queue size bytes.

Modifications

Add a new configuration pulsar.max-split-queue-cache-size to control the entry queue size bytes and the message queue size bytes. Half of this configuration will assign to entry queue size bytes and the left quota assign to message queue size bytes.

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

sijie · 2021-02-22T08:37:28Z

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/CacheSizeAllocator.java

+public class CacheSizeAllocator {
+
+    private final long maxCacheSize;
+    private final AtomicLong availableCacheSize;


Use LongAdder?

sijie · 2021-02-22T08:39:11Z

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java

@@ -136,6 +137,10 @@ public PulsarRecordCursor(List<PulsarColumnHandle> columnHandles, PulsarSplit pu
                        pulsarConnectorConfig),
                new PulsarConnectorMetricsTracker(pulsarConnectorCache.getStatsProvider()));
        this.decoderFactory = decoderFactory;
+        if (pulsarConnectorConfig.getMaxSplitEntryQueueSizeBytes() >= 0) {
+            this.entryCacheSizeAllocator = new CacheSizeAllocator(


I think you can make it an interface. Then have a no-op implementation and a CacheSizeAllocator implementation. Then you don't need to add if (entryCacheSizeAllocator != null) everywhere else.

Yes, that seems better.

codelipenghui

Please add unit tests for the changes to make sure the cache size allocator is initialized as expected.

codelipenghui · 2021-02-24T00:05:30Z

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java

@@ -136,6 +137,10 @@ public PulsarRecordCursor(List<PulsarColumnHandle> columnHandles, PulsarSplit pu
                        pulsarConnectorConfig),
                new PulsarConnectorMetricsTracker(pulsarConnectorCache.getStatsProvider()));
        this.decoderFactory = decoderFactory;
+        if (pulsarConnectorConfig.getMaxSplitEntryQueueSizeBytes() >= 0) {


If the size == 0, this means disabled?

If the config value is 0, there will be only one entry in the entry queue.

codelipenghui · 2021-02-26T00:56:58Z

@gaoran10 Could you please check the comments?

gaoran10 · 2021-03-02T08:07:55Z

/pulsarbot run-failure-checks

sijie · 2021-03-02T20:46:44Z

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java

@@ -222,7 +228,7 @@ public void setPulsarSqlSchemaInfoProvider(PulsarSqlSchemaInfoProvider schemaInf
    @VisibleForTesting
    class DeserializeEntries implements Runnable {

-    protected boolean isRunning = false;
+        protected boolean isRunning = false;


is this introduced by mistake?

This is an indent fix.

sijie · 2021-03-02T20:47:29Z

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java

@@ -654,4 +671,15 @@ private void checkFieldType(int field, Class<?> expected) {
        checkArgument(actual == expected, "Expected field %s to be type %s but is %s", field, expected, actual);
    }

+    private void initEntryCacheSizeAllocator(PulsarConnectorConfig connectorConfig) {
+        log.info("Init entry cache size allocator with max split entry queue size bytes {}.",


Move this log message to the if block?

sijie · 2021-03-02T20:47:58Z

...resto-pulsar/src/main/java/org/apache/pulsar/sql/presto/util/NoStrictCacheSizeAllocator.java

+
+    public long getAvailableCacheSize() {
+        try {
+            lock.readLock().lock();


Why do we need a lock?

sijie · 2021-03-02T20:49:38Z

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java

+                                cursor.asyncReadEntries(batchSize, maxSizeBytes,
+                                        this, System.nanoTime(), PositionImpl.latest);
+                            } else {
+                                metricsTracker.incr_READ_ATTEMPTS_FAIL();


If people doesn't configure this setting, we are initializing a NullCacheSizeAllocator. So it will cause metricsTracker to increase READ_ATTEMPTS_FAIL again and again.

Oh yes, I'll fix this.

codelipenghui · 2021-03-04T00:36:27Z

@gaoran10 Could you please check sijie's comments?

gaoran10 · 2021-03-08T07:52:34Z

/pulsarbot run-failure-checks

…zeAllocator` for invalid value of the configuration `maxSplitEntryQueueSizeBytes`.

gaoran10 · 2021-03-09T02:43:18Z

/pulsarbot run-failure-checks

…9628) ### Motivation In Pulsar SQL, there are two configurations `pulsar.max-split-entry-queue-size` and `pulsar.max-split-message-queue-size` to control the entry queue and message queue capacity, but some entries are so big some are small, it's hard to control the queue size bytes and the message queue size bytes. ### Modifications Add a new configuration `pulsar.max-split-queue-cache-size` to control the entry queue size bytes and the message queue size bytes. Half of this configuration will assign to entry queue size bytes and the left quota assign to message queue size bytes.

gaoran10 changed the title ~~[WIP] [Pulsar SQL] Add max split entry queue size bytes limitation~~ [Pulsar SQL] Add max split entry queue size bytes limitation Feb 19, 2021

merlimat requested a review from jerrypeng February 20, 2021 06:00

sijie reviewed Feb 22, 2021

View reviewed changes

codelipenghui assigned gaoran10 Feb 24, 2021

codelipenghui added the area/sql Pulsar SQL related features label Feb 24, 2021

codelipenghui added this to the 2.8.0 milestone Feb 24, 2021

codelipenghui reviewed Feb 24, 2021

View reviewed changes

gaoran10 force-pushed the pulsar-sql-read-cache-size-limit branch from 39dc7e6 to 1c4051d Compare March 2, 2021 03:49

sijie reviewed Mar 2, 2021

View reviewed changes

gaoran10 added 6 commits March 9, 2021 08:04

add max split entry queue size bytes limitation

26aa725

add the interface CacheSizeAllocator and add the class `NullCacheSi…

d464db5

…zeAllocator` for invalid value of the configuration `maxSplitEntryQueueSizeBytes`.

add unit test for cache size allocator

2a71da7

fix test

9aec766

add message queue cache size control

4f7c85b

adjust the configuration name

8e19763

gaoran10 force-pushed the pulsar-sql-read-cache-size-limit branch from 94d62ac to 8e19763 Compare March 9, 2021 00:05

sijie approved these changes Mar 13, 2021

View reviewed changes

codelipenghui approved these changes Mar 14, 2021

View reviewed changes

codelipenghui merged commit 1f6ce7a into apache:master Mar 14, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Pulsar SQL] Add max split entry queue size bytes limitation #9628

[Pulsar SQL] Add max split entry queue size bytes limitation #9628

gaoran10 commented Feb 19, 2021 •

edited

Loading

sijie Feb 22, 2021

gaoran10 Feb 28, 2021

sijie Feb 22, 2021

gaoran10 Feb 22, 2021

codelipenghui left a comment

codelipenghui Feb 24, 2021

gaoran10 Feb 28, 2021

codelipenghui commented Feb 26, 2021

gaoran10 commented Mar 2, 2021

sijie Mar 2, 2021

gaoran10 Mar 4, 2021

sijie Mar 2, 2021

sijie Mar 2, 2021

sijie Mar 2, 2021

gaoran10 Mar 4, 2021

codelipenghui commented Mar 4, 2021

gaoran10 commented Mar 8, 2021

gaoran10 commented Mar 9, 2021

[Pulsar SQL] Add max split entry queue size bytes limitation #9628

[Pulsar SQL] Add max split entry queue size bytes limitation #9628

Conversation

gaoran10 commented Feb 19, 2021 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Feb 26, 2021

gaoran10 commented Mar 2, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Mar 4, 2021

gaoran10 commented Mar 8, 2021

gaoran10 commented Mar 9, 2021

gaoran10 commented Feb 19, 2021 •

edited

Loading