You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2025/08/01 06:43:50 UTC

[PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleWithFixedRate [pulsar]

poorbarcode opened a new pull request, #24596:
URL: https://github.com/apache/pulsar/pull/24596

   ### Motivation
   
   **Background of `scheduleWithFixedDelay` and `scheduleWithFixedRate`**
   - `scheduleWithFixedRate`: Executes tasks at fixed intervals (period) regardless of task execution time, 
     - If task execution time ≤ period: Runs normally at the specified interval (e.g., every 1 second).
     - If task execution time > period: New tasks will trigger immediately (or queue up), causing backlog (may lead to consecutive executions). 
   - `scheduleWithFixedDelay`: Trigger Mechanism: Waits for the previous task to complete, then delays by the fixed interval (delay) before the next execution.
     - Regardless of task duration, the interval is always measured from the end of the last task.
   
   **Issue: broker ran checkMessageExpiry many times in 1s**
   
   When system resources are tight, it leads to a large backlog of tasks `checkMessageExpiry`, which are executed multiple times at the same time, causing even greater pressure
   
   ![image](https://github.com/user-attachments/assets/443df028-fb4d-4691-803b-efbdc46cf0b4)
   
   
   ### Modifications
   
   Call `scheduleWithFixedDelay` instead of `scheduleWithFixedRate`
   
   ### Documentation
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   - [ ] `doc` <!-- Your PR contains doc changes. -->
   - [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
   - [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
   - [ ] `doc-complete` <!-- Docs have been already added -->
   
   ### Matching PR in forked repository
   
   PR in forked repository: x


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleAtFixedRateNonConcurrently for scheduled tasks, instead of scheduleAtFixedRate [pulsar]

Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3273252804

   > Modifications
   > Call scheduleWithFixedDelay instead of scheduleWithFixedRate
   
   Please update the change description.


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2259484449


##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   I'd recommend implementing the wrapper for preventing concurrent execution of the `Runnable` task in the existing `org.apache.pulsar.common.util.Runnables` class, since in that case, it could be reused even without using this class. 



##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   The name `SingleThreadSafeScheduledExecutorService` is misleading. This is "single threaded", but the purpose is to prevent concurrent execution of tasks that are scheduled with fixed rate. Therefore a better name would be SingleThreadNonConcurrentFixedRateScheduler`. There's no need to extend `ScheduledThreadPoolExecutor`. This class could simply contain the reference to the `OrderedScheduler` and contain a single method to schedule tasks that are wrapped with a `Runnable` implementation that prevents concurrent execution.
   
   ```
   OrderedScheduler.newSchedulerBuilder()
                   .name(name)
                   .numThreads(1)
                   .build();
   ```
   
   There could be a single method `scheduleAtFixedRateNonConcurrently`.
   
   



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2259899284


##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   > I don't think that implementing a complete ScheduledExecutorService is necessary at all. The implementation looks very complex, which isn't needed to address the problem.
   
   If you ever read the implementation of `ScheduledThreadPoolExecutor` and the internal class `ScheduledFutureTask`, you will know that the new class is very simple, which just changed the implementation of get next runtime of the scheduled task: `setNextRunTime`
   
   >  I think that we shouldn't implement the ScheduledExecutorService at all and instead make it specialized for the use case where a single task is executed with an ordinary java.lang.Thread (well it makes sense to use Netty's io.netty.util.concurrent.FastThreadLocalThread thread implementation) and with logic which calculates the Thread.sleep between task executions so that it executes at fixed rate, preventing the backlogging of the single task that it can execute.
   
   The current implementation does this



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3150845380

   Even with this change, it's nearly impossible to prevent new code from calling `scheduleAtFixedRate`. It's unfair to say one is definitely better than the other.


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2259899284


##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   > I don't think that implementing a complete ScheduledExecutorService is necessary at all. The implementation looks very complex, which isn't needed to address the problem.
   
   If you ever read the implementation of `ScheduledThreadPoolExecutor` and the internal class `ScheduledFutureTask`, you will know that the new class is very simple, which just changed the implementation of get next runtime of the scheduled task: [setNextRunTime](https://github.com/apache/pulsar/pull/24596/files#diff-cc1fdb222ce13e343daa9172f667282c9fa0a96ade66830b87ff4066fc2d646aR345-R353)
   
   - the previous: `nextRunTime = latestRunTime + period`
   - the previous: `nextRunTime = latestRunTime + ((System.nanoTime() - latestRunTime) / period + 1) * period`
   
   >  I think that we shouldn't implement the ScheduledExecutorService at all and instead make it specialized for the use case where a single task is executed with an ordinary java.lang.Thread (well it makes sense to use Netty's io.netty.util.concurrent.FastThreadLocalThread thread implementation) and with logic which calculates the Thread.sleep between task executions so that it executes at fixed rate, preventing the backlogging of the single task that it can execute.
   
   The current implementation does this



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3153125921

   Actually the original analysis includes that point, but it's too long that I didn't include it.
   
   However, back to the issue, users should not configure a short interval like 1s and it's impossible because it can only be configured by `messageExpiryCheckIntervalInMinutes`, whose time unit is minute. Even with the smallest config (1 minute), could `scheduleAtFixedRate` be an actual issue? If so, it means `BrokerService#checkMessageExpiry` could take more than 1 minute, which is crazy. Instead of switching to `scheduleWithFixedDelay`, we should investigate why it could take so long. If it cannot be optimized, it should not be fixed at Pulsar side. In this case, it should be an operation issue that should be fixed by:
   - Increase the CPU (since it means there are a lot of topics to be served)
   - Increase the `messageExpiryCheckIntervalInMinutes` config (to avoid frequent time consuming operation)
   
   P.S. From the screenshot, message expiry check is scheduled every 5 minutes, which is a default value. It also shows no queued tasks being executed immediately.
   
   `scheduleWithFixedDelay` might be able to avoid queuing scheduled tasks in extreme cases, but from all existing info I can see, it does not solve an actual issue.


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2259899284


##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   > I don't think that implementing a complete ScheduledExecutorService is necessary at all. The implementation looks very complex, which isn't needed to address the problem.
   
   If you ever read the implementation of `ScheduledThreadPoolExecutor` and the internal class `ScheduledFutureTask`, you will know that the new class is very simple, which just changed the implementation of get next runtime of the scheduled task: [setNextRunTime](https://github.com/apache/pulsar/pull/24596/files#diff-cc1fdb222ce13e343daa9172f667282c9fa0a96ade66830b87ff4066fc2d646aR345-R353)
   
   - the previous: `nextRunTime = latestRunTime + period`
   - the PR did: `nextRunTime = latestRunTime + ((System.nanoTime() - latestRunTime) / period + 1) * period`
   
   >  I think that we shouldn't implement the ScheduledExecutorService at all and instead make it specialized for the use case where a single task is executed with an ordinary java.lang.Thread (well it makes sense to use Netty's io.netty.util.concurrent.FastThreadLocalThread thread implementation) and with logic which calculates the Thread.sleep between task executions so that it executes at fixed rate, preventing the backlogging of the single task that it can execute.
   
   The current implementation does this



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3150179414

   @poorbarcode The description refers `scheduleWithFixedRate`. The correct method name is `scheduleAtFixedRate`


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2259529184


##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   Renamed the class and the method
   - `SingleThreadSafeScheduledExecutorService` -> `SingleThreadNonConcurrentFixedRateScheduler`
   - `scheduleAtFixedRateAndDropOutdatedTask` -> `scheduleAtFixedRateNonConcurrently`



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleAtFixedRateNonConcurrently for scheduled tasks, instead of scheduleAtFixedRate [pulsar]

Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3326632647

   @lhotari please review this pr again


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2259537163


##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   > I'd recommend implementing the wrapper for preventing concurrent execution of the Runnable task in the existing org.apache.pulsar.common.util.Runnables class, since in that case, it could be reused even without using this class.
   
   I think the new class can benefit clearer, simpler and more efficient. BTW, we can hardly know whether the Runnble is convectively executing. 



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2259899284


##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   > I don't think that implementing a complete ScheduledExecutorService is necessary at all. The implementation looks very complex, which isn't needed to address the problem.
   
   If you ever read the implementation of `ScheduledThreadPoolExecutor` and the internal class `ScheduledFutureTask`, you will know that the new class is very simple, which just changes the implementation of get next runtime: `setNextRunTime`
   
   >  I think that we shouldn't implement the ScheduledExecutorService at all and instead make it specialized for the use case where a single task is executed with an ordinary java.lang.Thread (well it makes sense to use Netty's io.netty.util.concurrent.FastThreadLocalThread thread implementation) and with logic which calculates the Thread.sleep between task executions so that it executes at fixed rate, preventing the backlogging of the single task that it can execute.
   
   The current implementation does this



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleWithFixedRate [pulsar]

Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3143710351

   Could we just fix the `checkMessageExpiry ` from `scheduleWithFixedRate ` to `scheduleWithFixedDelay `?


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleWithFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2250774574


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1404,13 +1404,13 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
 
                             if (consumerFuture.complete(consumer)) {
                                 log.info("[{}] Created subscription on topic {} / {}",
-                                        remoteAddress, topicName, subscriptionName);
+                                        remoteAddress, topicName, subscriptionName); // TODO consumer id.

Review Comment:
   Please omit adding unrelated TODOs in this PR. You can keep the TODOs to yourself and handle these in a separate PR.



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3161021711

   >> Yes, adding a wrapper of scheduleAtFixedRate makes sense
   >> That's why I'd recommend using scheduleAtFixedRate with the wrapper that optionally logs a warning when the previous execution is still running.
   
   > @BewareMyPower @lhotari Thanks for your suggestion, I will improve the implementation 
   
   Finished, please review again


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleAtFixedRateNonConcurrently for scheduled tasks, instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3277223534

   @coderzc 
   
   > Please update the change description.
   
   Thanks for mentioning this, please review again


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleWithFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2250994027


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1404,13 +1404,13 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
 
                             if (consumerFuture.complete(consumer)) {
                                 log.info("[{}] Created subscription on topic {} / {}",
-                                        remoteAddress, topicName, subscriptionName);
+                                        remoteAddress, topicName, subscriptionName); // TODO consumer id.

Review Comment:
   Thanks for mentioning this, I have removed these changes



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleWithFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2250774574


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1404,13 +1404,13 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
 
                             if (consumerFuture.complete(consumer)) {
                                 log.info("[{}] Created subscription on topic {} / {}",
-                                        remoteAddress, topicName, subscriptionName);
+                                        remoteAddress, topicName, subscriptionName); // TODO consumer id.

Review Comment:
   Please omit adding TODOs in this PR. You can keep the TODOs to yourself and handle these in a separate PR.



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleWithFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2251167322


##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java:
##########
@@ -231,7 +231,7 @@ private ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore,
         this.entryCacheManager = new RangeEntryCacheManagerImpl(this, scheduledExecutor, openTelemetry);
         this.statsTask = scheduledExecutor.scheduleWithFixedDelay(catchingAndLoggingThrowables(this::refreshStats),
                 0, config.getStatsPeriodSeconds(), TimeUnit.SECONDS);
-        this.flushCursorsTask = scheduledExecutor.scheduleAtFixedRate(catchingAndLoggingThrowables(this::flushCursors),
+        this.flushCursorsTask = scheduledExecutor.scheduleWithFixedDelay(catchingAndLoggingThrowables(this::flushCursors),
                 config.getCursorPositionFlushSeconds(), config.getCursorPositionFlushSeconds(), TimeUnit.SECONDS);

Review Comment:
   Changing from `scheduleAtFixedRate` to `scheduleWithFixedDelay` could have a significant impact to behavior. Let's say if cursorPositionFlushSeconds is set to 1 and flushing takes 300ms. After the change, cursors would get flushed every 1300 ms instead of every 1000 ms.
   This is why I think that we should consider each case closely where we are going to make a change from `scheduleAtFixedRate` to `scheduleWithFixedDelay`.
   One possible alternative solution would be to keep on using `scheduleAtFixedRate` and implement the wrapper in `Runnables` that I explained in a separate comment.
   



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3153738905

   Yes, adding a wrapper of `scheduleAtFixedRate` makes sense


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3163394269

   ## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24596?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :x: Patch coverage is `71.65354%` with `36 lines` in your changes missing coverage. Please review.
   :white_check_mark: Project coverage is 74.30%. Comparing base ([`bbc6224`](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`da0c2be`](https://app.codecov.io/gh/apache/pulsar/commit/da0c2becfc268555a3dbf098994c850af083ddbe?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   :warning: Report is 1254 commits behind head on master.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24596?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [...l/SingleThreadNonConcurrentFixedRateScheduler.java](https://app.codecov.io/gh/apache/pulsar/pull/24596?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Futil%2FSingleThreadNonConcurrentFixedRateScheduler.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi91dGlsL1NpbmdsZVRocmVhZE5vbkNvbmN1cnJlbnRGaXhlZFJhdGVTY2hlZHVsZXIuamF2YQ==) | 67.28% | [20 Missing and 15 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24596?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [...rg/apache/pulsar/broker/service/BrokerService.java](https://app.codecov.io/gh/apache/pulsar/pull/24596?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FBrokerService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0Jyb2tlclNlcnZpY2UuamF2YQ==) | 95.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24596?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   [![Impacted file tree graph](https://app.codecov.io/gh/apache/pulsar/pull/24596/graphs/tree.svg?width=650&height=150&src=pr&token=acYqCpsK9J&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)](https://app.codecov.io/gh/apache/pulsar/pull/24596?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #24596      +/-   ##
   ============================================
   + Coverage     73.57%   74.30%   +0.73%     
   - Complexity    32624    33185     +561     
   ============================================
     Files          1877     1882       +5     
     Lines        139502   146862    +7360     
     Branches      15299    16875    +1576     
   ============================================
   + Hits         102638   109133    +6495     
   - Misses        28908    29052     +144     
   - Partials       7956     8677     +721     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24596/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.59% <53.54%> (+2.01%)` | :arrow_up: |
   | [systests](https://app.codecov.io/gh/apache/pulsar/pull/24596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.29% <53.54%> (-1.03%)` | :arrow_down: |
   | [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24596/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.79% <71.65%> (+0.95%)` | :arrow_up: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24596?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [...rg/apache/pulsar/broker/service/BrokerService.java](https://app.codecov.io/gh/apache/pulsar/pull/24596?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FBrokerService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0Jyb2tlclNlcnZpY2UuamF2YQ==) | `83.63% <95.00%> (+2.85%)` | :arrow_up: |
   | [...l/SingleThreadNonConcurrentFixedRateScheduler.java](https://app.codecov.io/gh/apache/pulsar/pull/24596?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Futil%2FSingleThreadNonConcurrentFixedRateScheduler.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi91dGlsL1NpbmdsZVRocmVhZE5vbkNvbmN1cnJlbnRGaXhlZFJhdGVTY2hlZHVsZXIuamF2YQ==) | `67.28% <67.28%> (ø)` | |
   
   ... and [1125 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24596/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   </details>
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleWithFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3149871991

   > I think maybe we don't need to replace all `scheduleWithFixedDelay` to `scheduleWithFixedRate`, for these tasks which costs long time, replace to `scheduleWithFixedRate` I afraid it will cost high broker work load?
   
   @dao-jun  I don't think that anyone is suggesting to replace `scheduleWithFixedDelay` with `scheduleWithFixedRate`. It's the other way around that is in this PR.


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleWithFixedRate [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3149853114

   I think maybe we don't need to replace all `scheduleWithFixedDelay` to `scheduleWithFixedRate`, for these tasks which costs long time, replace to `scheduleWithFixedRate` I afraid it will cost high broker work load?


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3151156922

   > This PR involves many changes.
   > 
   > Share the comparison of these two methods from Gemini AI:
   > 
   > > `scheduleAtFixedRate` is best for:
   > > Time-sensitive tasks that need a fixed schedule (e.g., cron jobs, hourly chimes).	Tasks where a fixed pause between executions is important, especially if task duration is unpredictable.
   > > `scheduleWithFixedDelay` is best for
   > > Tasks where a fixed pause between executions is important, especially if task duration is unpredictable.
   > > The choice between scheduleAtFixedRate and scheduleWithFixedDelay depends entirely on your specific use case.
   > 
   > Since the issue is exactly with `checkMessageExpiry`, please only change its related logic.
   
   @BewareMyPower I think that analysis misses the detail that `scheduleWithFixedDelay` will prevent concurrent execution. My suggestion would be to continue to use `scheduleAtFixedRate`, but add the solution to `Runnables` class to prevent concurrent execution which could cause many issues. (explained in my previous comment)


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleAtFixedRateNonConcurrently for scheduled tasks, instead of scheduleAtFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #24596:
URL: https://github.com/apache/pulsar/pull/24596


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3162706551

   /pulsarbot rerun-failure-checks


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3153720843

   > `scheduleWithFixedDelay` might be able to avoid queuing scheduled tasks in extreme cases, but from all existing info I can see, it does not solve an actual issue.
   
   @BewareMyPower I agree that it doesn't fix the actual issue. That's why I'd recommend using `scheduleAtFixedRate` with the wrapper that optionally logs a warning when the previous execution is still running. In that case, it could be caused by a deadlock, for example. Logging would help in investigating and fixing the root cause. It would also be possible to track the thread that is in execution and log the stack trace of the thread that is stuck in executing the task.
   
   Without preventing concurrent job execution, the problem would get worse since another task would start executing the same task concurrently. That could already happen in cases where we have very short delays and lets say use synchronization to run the task one-by-one. All later coming tasks would get back logged in the executor which is a problem that should be solved. I have understood that it's the problem that @poorbarcode would like to address.


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#issuecomment-3155650128

   > Yes, adding a wrapper of scheduleAtFixedRate makes sense
   
   @BewareMyPower @lhotari Thanks for your suggestion, I will improve the implementation 
   


-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [improve][broker]Call scheduleWithFixedDelay instead of scheduleAtFixedRate [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24596:
URL: https://github.com/apache/pulsar/pull/24596#discussion_r2259876043


##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/SingleThreadSafeScheduledExecutorService.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableScheduledFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SingleThreadSafeScheduledExecutorService extends ScheduledThreadPoolExecutor

Review Comment:
   > I think the new class can benefit clearer, simpler and more efficient. 
   
   I don't think that implementing a complete ScheduledExecutorService is necessary at all. The implementation looks very complex, which isn't needed to address the problem.
   
   > BTW, we can hardly know whether the Runnble is convectively executing.
   
   Oh yes, now I see the problem. With a single threaded executor, there wouldn't be a way to skip scheduling the task if the previous task is overdue. 
   
   For a multi-threaded task executor, this type of wrapper would be useful:
   
   ```java
       public static Runnable preventConcurrentExecution(Runnable runnable, String taskName) {
           return new NonConcurrentRunnable(runnable, taskName);
       }
   
       private static final class NonConcurrentRunnable implements Runnable {
           private final Runnable runnable;
           private final String taskName;
           private final AtomicLong taskStarted = new AtomicLong(0);
   
           private NonConcurrentRunnable(Runnable runnable, String taskName) {
               this.runnable = runnable;
               this.taskName = taskName;
           }
   
           @Override
           public void run() {
               if (taskStarted.compareAndSet(0L, System.nanoTime())) {
                   try {
                       runnable.run();
                   } catch (Throwable t) {
                       LOGGER.error("Unexpected throwable caught when executing task '{}'", taskName, t);
                   } finally {
                       taskStarted.set(0L);
                   }
               } else {
                   long elapsedTimeNanos = System.nanoTime() - taskStarted.get();
                   LOGGER.warn(
                           "Runnable for task '{}' is already running, skipping execution. "
                           + "Elapsed time since current task started: {} ms", taskName,
                           TimeUnit.NANOSECONDS.toMillis(elapsedTimeNanos));
               }
           }
       }
   ```
   
   In this case, since we are executing in a single dedicated thread, I think that we shouldn't implement the ScheduledExecutorService at all and instead make it specialized for the use case where a single task is executed with an ordinary java.lang.Thread (well it makes sense to use Netty's io.netty.util.concurrent.FastThreadLocalThread thread implementation) and with logic which calculates the Thread.sleep between task executions so that it executes at fixed rate, preventing the backlogging of the single task that it can execute.



-- 
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: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org