You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "suneet-s (via GitHub)" <gi...@apache.org> on 2023/02/27 00:25:17 UTC
[GitHub] [druid] suneet-s opened a new pull request, #13852: Continuous automatic compaction
suneet-s opened a new pull request, #13852:
URL: https://github.com/apache/druid/pull/13852
### Description
This change introduces the ability to have auto-compaction continuously
schedule compaction tasks as slots become available.
Previously, each run of the CompactSegments duty built an iterator based on
the latest segment metadata available. This meant that if the compact tasks
that were scheduled ran into any issues, like task lock contention, or an
interval which can not be compacted because of a bug, auto-compaction would
be stuck on the cluster.
With this change, CompactSegments, refreshes it's view of the segments based
on the `druid.coordinator.compaction.searchPolicyRefreshPeriod` property.
This allows auto-compaction to continue to make progress if any interval
fails to compact until the search policy is refreshed.
Compaction statistics for the cluster are only refreshed when the search
policy is refreshed. This is because to collect statistics, the task has
to run through the entire list of available segments on the cluster which
can take a long time on large clusters
To enable this behavior on the cluster, add something like this to the
coordinator runtime properties
```
druid.coordinator.dutyGroups=["compaction"]
druid.coordinator.compaction.duties=["compactSegments"]
druid.coordinator.compaction.period=PT60S
```
#### Release note
New: You can now schedule automatic compaction to run continuously and configure
how frequently it should consider new segments for compaction
<hr>
##### Key changed/added classes in this PR
* CompactionSegmentSearchPolicy#resetIfNeeded
* CompactSegments#makeStats
<hr>
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] imply-cheddar commented on pull request #13852: Continuous automatic compaction
Posted by "imply-cheddar (via GitHub)" <gi...@apache.org>.
imply-cheddar commented on PR #13852:
URL: https://github.com/apache/druid/pull/13852#issuecomment-1453044252
This change does not deserve a new config and we cannot merge it in a way that requires a new config. If we separate out to a queue that is read, then the Duty stays the same and populates that queue, the thread just runs stuff based on whatever is in that queue. It does not need an extra config and does not need any extra thought by end users. It will immediately benefit all deployments. Adding more configs is a very expensive operation and adding a new config that is hard to understand (why would we be setting the reset *longer* than something else?) is even worse, especially for functionality that will benefit all deployments and that we want on for everybody.
Please fix the PR to remove all new configs and make it just operate by having the duty mutate a queue (each time it wakes up) and have a new thread that reads from that queue and submits jobs in accordance with whatever configurations are set in terms of how many task slots can be used.
As it is, I'm -1 on this PR until these changes are made.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] clintropolis commented on a diff in pull request #13852: Continuous automatic compaction
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #13852:
URL: https://github.com/apache/druid/pull/13852#discussion_r1119618425
##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java:
##########
@@ -137,4 +137,7 @@ public boolean getCompactionSkipLockedIntervals()
return true;
}
+ @Config("druid.coordinator.compaction.searchPolicyRefreshPeriod")
+ @Default("PT5M")
Review Comment:
ah the name doesn't really bother me, it was more of a general comment about the config of this stuff being all over the place and that I have no idea where something should actually live.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Continuous automatic compaction (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13852:
URL: https://github.com/apache/druid/pull/13852#issuecomment-1933155051
This pull request has been marked as stale due to 60 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If you think
that's incorrect or this pull request should instead be reviewed, please simply
write any comment. Even if closed, you can still revive the PR at any time or
discuss it on the dev@druid.apache.org list.
Thank you for your contributions.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] clintropolis commented on a diff in pull request #13852: Continuous automatic compaction
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #13852:
URL: https://github.com/apache/druid/pull/13852#discussion_r1119616973
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstPolicy.java:
##########
@@ -20,34 +20,78 @@
package org.apache.druid.server.coordinator.duty;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.Interval;
+import java.time.Clock;
import java.util.List;
import java.util.Map;
/**
* This policy searches segments for compaction from the newest one to oldest one.
+ * The {@link #resetIfNeeded} functionality is inspired by {@link com.google.common.base.Suppliers.ExpiringMemoizingSupplier}.
*/
public class NewestSegmentFirstPolicy implements CompactionSegmentSearchPolicy
{
private final ObjectMapper objectMapper;
+ private final long durationMillis;
+ private transient volatile NewestSegmentFirstIterator iterator;
+ // The special value 0 means "not yet initialized".
+ private transient volatile long expirationMillis;
+ private final Clock clock;
@Inject
- public NewestSegmentFirstPolicy(ObjectMapper objectMapper)
+ public NewestSegmentFirstPolicy(ObjectMapper objectMapper, DruidCoordinatorConfig config, Clock clock)
{
this.objectMapper = objectMapper;
+ this.durationMillis = config.getCompactionSearchPolicyRefreshPeriod().getMillis();
+ this.clock = clock;
+ Preconditions.checkArgument(durationMillis > 0);
}
@Override
- public CompactionSegmentIterator reset(
+ public Pair<CompactionSegmentIterator, Boolean> resetIfNeeded(
Map<String, DataSourceCompactionConfig> compactionConfigs,
Map<String, SegmentTimeline> dataSources,
Map<String, List<Interval>> skipIntervals
)
{
- return new NewestSegmentFirstIterator(objectMapper, compactionConfigs, dataSources, skipIntervals);
+ long millis = expirationMillis;
+ long now = clock.millis();
+ if (millis == 0 || now - millis >= 0) {
+ synchronized (this) {
Review Comment:
oops, i can't read, sorry 🙃
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Continuous automatic compaction (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13852:
URL: https://github.com/apache/druid/pull/13852#issuecomment-1984824178
This pull request/issue has been closed due to lack of activity. If you think that
is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s commented on a diff in pull request #13852: Continuous automatic compaction
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on code in PR #13852:
URL: https://github.com/apache/druid/pull/13852#discussion_r1120641481
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstPolicy.java:
##########
@@ -20,34 +20,78 @@
package org.apache.druid.server.coordinator.duty;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.Interval;
+import java.time.Clock;
import java.util.List;
import java.util.Map;
/**
* This policy searches segments for compaction from the newest one to oldest one.
+ * The {@link #resetIfNeeded} functionality is inspired by {@link com.google.common.base.Suppliers.ExpiringMemoizingSupplier}.
*/
public class NewestSegmentFirstPolicy implements CompactionSegmentSearchPolicy
{
private final ObjectMapper objectMapper;
+ private final long durationMillis;
+ private transient volatile NewestSegmentFirstIterator iterator;
Review Comment:
sgtm I'll remove the transient to reduce confusion
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] clintropolis commented on a diff in pull request #13852: Continuous automatic compaction
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #13852:
URL: https://github.com/apache/druid/pull/13852#discussion_r1119618425
##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java:
##########
@@ -137,4 +137,7 @@ public boolean getCompactionSkipLockedIntervals()
return true;
}
+ @Config("druid.coordinator.compaction.searchPolicyRefreshPeriod")
+ @Default("PT5M")
Review Comment:
ah the name doesn't really bother me, it was more of a general comment about the config of this stuff being all over the place and that I have no idea where something should actually live. Given that, I think here is probably as fine as anywhere else 😛
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] github-code-scanning[bot] commented on a diff in pull request #13852: Continuous automatic compaction
Posted by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #13852:
URL: https://github.com/apache/druid/pull/13852#discussion_r1121195398
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java:
##########
@@ -450,7 +469,11 @@
"coordinator-issued",
segmentsToCompact,
config.getTaskPriority(),
- ClientCompactionTaskQueryTuningConfig.from(config.getTuningConfig(), config.getMaxRowsPerSegment(), config.getMetricsSpec() != null),
+ ClientCompactionTaskQueryTuningConfig.from(
+ config.getTuningConfig(),
+ config.getMaxRowsPerSegment(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [DataSourceCompactionConfig.getMaxRowsPerSegment](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/4366)
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Continuous automatic compaction (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #13852: Continuous automatic compaction
URL: https://github.com/apache/druid/pull/13852
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] clintropolis commented on a diff in pull request #13852: Continuous automatic compaction
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #13852:
URL: https://github.com/apache/druid/pull/13852#discussion_r1119621427
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstPolicy.java:
##########
@@ -20,34 +20,78 @@
package org.apache.druid.server.coordinator.duty;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.Interval;
+import java.time.Clock;
import java.util.List;
import java.util.Map;
/**
* This policy searches segments for compaction from the newest one to oldest one.
+ * The {@link #resetIfNeeded} functionality is inspired by {@link com.google.common.base.Suppliers.ExpiringMemoizingSupplier}.
*/
public class NewestSegmentFirstPolicy implements CompactionSegmentSearchPolicy
{
private final ObjectMapper objectMapper;
+ private final long durationMillis;
+ private transient volatile NewestSegmentFirstIterator iterator;
Review Comment:
I guess I agree that `transient` is technically harmless since we don't participate in `Serializable`, I was more thinking that it being there makes the reader wonder if we do or why its there (like i did), and since we don't use `transient` anywhere else in the code.
Guava has it because
```
static class MemoizingSupplier<T> implements Supplier<T>, Serializable
```
so it needs these things because it implements `Serializable`, but we do not.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s commented on pull request #13852: Continuous automatic compaction
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on PR #13852:
URL: https://github.com/apache/druid/pull/13852#issuecomment-1450147069
Thanks for the reviews @imply-cheddar and @kfaraz
> Another way you could achieve the intended outcome and, I think, have a bit more straightforward code (at least, you wouldn't need any new configs) would be to introduce a new thread.
I described this approach in the alternatives considered section of the PR. It does clean up the code a little bit (primarily that `CompactionSegmentSearchPolicy#resetIfNeeded` doesn't need to return a pair so callers know when the iterator was reset). However, we still end up with 2 configs - one for the period at which the iterator should be reset and another for the period at which compact tasks should be scheduled. The other reason I did not go with this approach is to preserve compatibility for users currently using the CompactSegments duty as a custom duty. Trying to disable the old way of running auto-compaction and enable 2 new duties to run it would make things more error prone imo.
> The CoordinatorSimulation is intended to allow you to create tests that simulate states of clusters.
Thanks for this. I did not know about this class, I will add some tests here.
> The other advantages of this approach are that users can benefit from it even if they don't specify a custom period for CompactSegments.
I have purposefully implemented this change in a way where there should be no user visible change yet. I would like to be able to enable this on a few production clusters and test this in real world scenarios before Druid users are opted in to this by default. A future patch should change the defaults so users benefit from this even if they make no configuration 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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] clintropolis commented on a diff in pull request #13852: Continuous automatic compaction
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #13852:
URL: https://github.com/apache/druid/pull/13852#discussion_r1119537778
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstPolicy.java:
##########
@@ -20,34 +20,78 @@
package org.apache.druid.server.coordinator.duty;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.Interval;
+import java.time.Clock;
import java.util.List;
import java.util.Map;
/**
* This policy searches segments for compaction from the newest one to oldest one.
+ * The {@link #resetIfNeeded} functionality is inspired by {@link com.google.common.base.Suppliers.ExpiringMemoizingSupplier}.
*/
public class NewestSegmentFirstPolicy implements CompactionSegmentSearchPolicy
{
private final ObjectMapper objectMapper;
+ private final long durationMillis;
+ private transient volatile NewestSegmentFirstIterator iterator;
Review Comment:
is `transient` really needed here? (tbh im not very familiar with this keyword... 😅 )
Afaict its related to `java.io.Serializable` which afaik we don't really use, but i could totally be wrong. I see barely any places using it in our codebase anyway...
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstPolicy.java:
##########
@@ -20,34 +20,78 @@
package org.apache.druid.server.coordinator.duty;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.Interval;
+import java.time.Clock;
import java.util.List;
import java.util.Map;
/**
* This policy searches segments for compaction from the newest one to oldest one.
+ * The {@link #resetIfNeeded} functionality is inspired by {@link com.google.common.base.Suppliers.ExpiringMemoizingSupplier}.
*/
public class NewestSegmentFirstPolicy implements CompactionSegmentSearchPolicy
{
private final ObjectMapper objectMapper;
+ private final long durationMillis;
+ private transient volatile NewestSegmentFirstIterator iterator;
+ // The special value 0 means "not yet initialized".
+ private transient volatile long expirationMillis;
+ private final Clock clock;
@Inject
- public NewestSegmentFirstPolicy(ObjectMapper objectMapper)
+ public NewestSegmentFirstPolicy(ObjectMapper objectMapper, DruidCoordinatorConfig config, Clock clock)
{
this.objectMapper = objectMapper;
+ this.durationMillis = config.getCompactionSearchPolicyRefreshPeriod().getMillis();
+ this.clock = clock;
+ Preconditions.checkArgument(durationMillis > 0);
}
@Override
- public CompactionSegmentIterator reset(
+ public Pair<CompactionSegmentIterator, Boolean> resetIfNeeded(
Map<String, DataSourceCompactionConfig> compactionConfigs,
Map<String, SegmentTimeline> dataSources,
Map<String, List<Interval>> skipIntervals
)
{
- return new NewestSegmentFirstIterator(objectMapper, compactionConfigs, dataSources, skipIntervals);
+ long millis = expirationMillis;
+ long now = clock.millis();
+ if (millis == 0 || now - millis >= 0) {
+ synchronized (this) {
Review Comment:
why does this need synchronized (or why doesn't `reset` need synchronized?)
##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java:
##########
@@ -137,4 +137,7 @@ public boolean getCompactionSkipLockedIntervals()
return true;
}
+ @Config("druid.coordinator.compaction.searchPolicyRefreshPeriod")
+ @Default("PT5M")
Review Comment:
curious, why here instead of on the `CompactSegments` duty? I guess that would make the property `druid.coordinator.compaction.duty.searchPolicyRefreshPeriod` instead
side note, coordinator config in general seems really complicated 😅 I had to read a bunch of code to understand how custom duties work and get wired up to stuff... and its kind of strange.
I guess where I am getting at is that it seems like having this refresh period be more frequent than the duty period seems like it would be an incorrect configuration (or at least useless since it would always reset), but I'm not entirely sure how such a check could actually be wired up. Maybe if the duty period was added to the properties that get injected so the compaction duty could pick it up or something?
There is also `CoordinatorCompactionConfig` to add to the confusion... not to mention `druid.coordinator.kill.compaction.period` which _does_ live here...
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] kfaraz commented on pull request #13852: Continuous automatic compaction
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #13852:
URL: https://github.com/apache/druid/pull/13852#issuecomment-1449814414
Thanks a lot for taking this up, @suneet-s !
I had a similar approach in mind to what @imply-cheddar has described.
@suneet-s , I think you too have mentioned this as a design that you considered, breaking up the `CompactSegments` duty into two.
I feel it would be a cleaner approach and would avoid the need to add a new config.
- Keep the `CompactSegments` duty mostly unchanged
- it would run at the custom period, if specified, otherwise, the default `indexingPeriod` (30 min)
- it continues to call `policy.reset()` to create a new iterator
- pass the `CompactionJobQueue` to this duty and update it with the new iterator
- Optional: Rename this duty to reflect that it just identifies compactible intervals (maybe `GetIntervalsForCompaction` or something) and does not actually invoke compaction.
- Add a new duty `QueueCompactionTasks`
- This runs as frequently as other coordinator duties (maybe alongwith historical management duties, i.e. every 1 min or so).
- Avoids the need of having a separate executor/thread and is frequent enough to make the best use of available compaction task slots.
- Asks the `CompactionJobQueue` for the next compaction tasks to queue and sends them to the Overlord
- For the most part, this translates to moving the `CompactSegments.doRun()` method to this new duty
- Stats computation
- It can remain in `CompactSegments` if it is a costly operation
- If required, we could even choose to expand `CompactionJobQueue` to be a `ClusterCompactionState` which could accumulate stats and snapshots as and when they are built and report them when necessary.
The other advantages of this approach are that users can benefit from it even if they don't specify a custom period for `CompactSegments`. (In the future, it might even help us do away with having `CompactSegments` as a custom duty, which is a little weird since compaction is a core feature of the ingestion system.)
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s commented on a diff in pull request #13852: Continuous automatic compaction
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on code in PR #13852:
URL: https://github.com/apache/druid/pull/13852#discussion_r1119598945
##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java:
##########
@@ -137,4 +137,7 @@ public boolean getCompactionSkipLockedIntervals()
return true;
}
+ @Config("druid.coordinator.compaction.searchPolicyRefreshPeriod")
+ @Default("PT5M")
Review Comment:
I had considered placing the config in the `CoordinatorCompactionConfig` instead, but decided against it because that config is dynamic, but changing that value would not necessarily update the expiry time of the iterator, so this seemed like a better UX
`druid.coordinator.compaction. searchPolicyRefreshPeriod` could use a better name. It's intent is to provide operators with the ability to choose when to re-build the view of the segment interval timeline to be considered for compaction. Today, every time the compact segments duty is run, the view is re-built. So the property does not need to be tied to the CompactSegments duty
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstPolicy.java:
##########
@@ -20,34 +20,78 @@
package org.apache.druid.server.coordinator.duty;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.Interval;
+import java.time.Clock;
import java.util.List;
import java.util.Map;
/**
* This policy searches segments for compaction from the newest one to oldest one.
+ * The {@link #resetIfNeeded} functionality is inspired by {@link com.google.common.base.Suppliers.ExpiringMemoizingSupplier}.
*/
public class NewestSegmentFirstPolicy implements CompactionSegmentSearchPolicy
{
private final ObjectMapper objectMapper;
+ private final long durationMillis;
+ private transient volatile NewestSegmentFirstIterator iterator;
+ // The special value 0 means "not yet initialized".
+ private transient volatile long expirationMillis;
+ private final Clock clock;
@Inject
- public NewestSegmentFirstPolicy(ObjectMapper objectMapper)
+ public NewestSegmentFirstPolicy(ObjectMapper objectMapper, DruidCoordinatorConfig config, Clock clock)
{
this.objectMapper = objectMapper;
+ this.durationMillis = config.getCompactionSearchPolicyRefreshPeriod().getMillis();
+ this.clock = clock;
+ Preconditions.checkArgument(durationMillis > 0);
}
@Override
- public CompactionSegmentIterator reset(
+ public Pair<CompactionSegmentIterator, Boolean> resetIfNeeded(
Map<String, DataSourceCompactionConfig> compactionConfigs,
Map<String, SegmentTimeline> dataSources,
Map<String, List<Interval>> skipIntervals
)
{
- return new NewestSegmentFirstIterator(objectMapper, compactionConfigs, dataSources, skipIntervals);
+ long millis = expirationMillis;
+ long now = clock.millis();
+ if (millis == 0 || now - millis >= 0) {
+ synchronized (this) {
Review Comment:
`reset` does have synchronized in the function definition.
I will add more comments explaining the thread safety aspects in the next patch. The need primarily arises from the `force compaction` API that is currently only marked safe for testing. With this change, it should be safe to call it in all scenarios.
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/NewestSegmentFirstPolicy.java:
##########
@@ -20,34 +20,78 @@
package org.apache.druid.server.coordinator.duty;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.Interval;
+import java.time.Clock;
import java.util.List;
import java.util.Map;
/**
* This policy searches segments for compaction from the newest one to oldest one.
+ * The {@link #resetIfNeeded} functionality is inspired by {@link com.google.common.base.Suppliers.ExpiringMemoizingSupplier}.
*/
public class NewestSegmentFirstPolicy implements CompactionSegmentSearchPolicy
{
private final ObjectMapper objectMapper;
+ private final long durationMillis;
+ private transient volatile NewestSegmentFirstIterator iterator;
Review Comment:
I think transient just means don't serialize this object because it is computed. This search policy shouldn't be serialized, but this pattern is used in Suppliers.memoizingSupplierWithExpiration, so I figured I'd stick with it since it is battle tested. I meant to add a comment providing credit for the inspiration of this pattern - I'll update the docs
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] paul-rogers commented on pull request #13852: Continuous automatic compaction
Posted by "paul-rogers (via GitHub)" <gi...@apache.org>.
paul-rogers commented on PR #13852:
URL: https://github.com/apache/druid/pull/13852#issuecomment-1455001687
I’m a newbie to this code so I took a deep dive to understand how it works. The compaction scheduling algorithm is rather crude: it isn’t well designed for a large, busy, distributed system. @suneet-s ’s fix is a worth-while short term fix, but this code really needs a redesign along the lines outlined by @imply-cheddar and @kfaraz.
For anyone else new to this area, here’s an overview of the logic. Please forgive any newbie misunderstandings. At the highest level:
* The coordinator provides a set of duties which are called one after another on a schedule.
* Duties have Guice-injected global state, but no local state across calls.
* Each duty is called from the main duty loop and thus must complete quickly.
* The compaction scheduler is a duty. As such, each call is independent from any previous call, and the scheduler must make its decisions quickly.
The challenge, then, is to design the compaction scheduler to be stateless and fast. On each call it:
* Determines the number of available slots, accounting for running tasks.
* Creates a big iterator over all datasources with compaction specs and their segments sorted by time chunk, minus any segments locked by existing compaction or ingestion tasks.
* Launches a set of compaction tasks from the head of the iterator until slots are full.
In short, each call to the scheduler duty builds a large list of candidate actions: “an iterator based on the latest segment metadata available”. The scheduler then starts working down that list from the head, stopping when the worker slots are filled. If “the compact tasks that were scheduled ran into any issues, like task lock contention, or an interval which can not be compacted because of a bug” then the next invocation of scheduler will launch new tasks for those same datasources and time chunks. It will do so because the scheduler maintains no state that would tell it that those very time chunks just failed. The new tasks may also fail. The result is “auto-compaction would be stuck on the cluster.”
The design works well if the number of datasource and segments is small relative to the number of available compaction tasks. That assumption is not valid on larger systems.
The design also assumes that new ingestion tasks will seldom cancel compaction tasks, hence a start-from-the-top approach will make progress. That assumption is not valid on a system with late-arriving data. Yet, late-arriving data is a fact of life in many systems.
Again, this code needs a redesign. But, that is a major project. So, what short-term solution can we apply instead?
The obvious solution is to allow the list (the iterator) to persist, and to have the scheduler work its way through the entire iterator before starting over again. Doing so requires the scheduler to maintain state.
Suneet's fix places the iterator state not on the scheduler itself, but rather on a `CompactionSegmentSearchPolicy` instance, which is a bit of a back-door solution: the coordinator duty does not maintain state, but a Guice-injected global dependency does.
The original code built the iterator anew on each scheduler invocation. The new code hedges its bets: it does not build the iterator on each call, but rather after some amount of time. There is much discussion about the configuration of this refresh period. No doubt it would be quite hard for anyone to come up with a good number.
Perhaps one simple solution is to omit the refresh period. Instead, the scheduler works its way through one entire list (iterator) before looping back to the start. Let's call this "one cycle". The full-cycle approach ensures all time chunks have a shot at compaction. If a task fails during one cycle, that chunk will be tried again in the next cycle, perhaps after any active ingestion has completed. Failures in compaction will occur once per cycle, not once per scheduler invocation. Compaction will no longer become "stuck."
A risk, of course, is that the information in the iterator becomes stale. The system is distributed: it should already handle that case: race conditions are to be expected. Another risk is that new data arrives faster than a cycle can run. In this case, no algorithm can solve the problem: more compaction resources are the only answer.
So, one question is, would the full-cycle idea work in practice?
The current PR is clearly an improvement. A redesign, as outlined by @imply-cheddar and @kfaraz would be a project, made more of a challenge because tests in this area appear to be sparse. The code is essentially untestable without a cluster because it is non-modular and is tightly coupled with other parts of the system. So, a second question is: can we accept this short-term fix or would we rather wait indefinitely for someone to tackle the sorely-needed redesign?
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s commented on pull request #13852: Continuous automatic compaction
Posted by "suneet-s (via GitHub)" <gi...@apache.org>.
suneet-s commented on PR #13852:
URL: https://github.com/apache/druid/pull/13852#issuecomment-1450154390
I really want to be able to start iterating on this and pushing up more patches to expand on the functionality. Other than the comment about tests (which i strongly agree with and can address shortly) are any of these comments meant to block the PR from being merged?
I haven't added docs to this PR, because I don't want users to know about this capability yet. Some things like the config might change in future patches (based on the feedback in this PR and using it in real production clusters), but I believe as written, there is no user visible change to users unless they know about this setting.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] imply-cheddar commented on pull request #13852: Continuous automatic compaction
Posted by "imply-cheddar (via GitHub)" <gi...@apache.org>.
imply-cheddar commented on PR #13852:
URL: https://github.com/apache/druid/pull/13852#issuecomment-1449376685
Another way you could achieve the intended outcome and, I think, have a bit more straightforward code (at least, you wouldn't need any new configs) would be to introduce a new thread. Basically, continue to use the duty to figure out which time chunks should be compacted by looking at the segment list/whatever. The duty populates the state of some object, let's call it a `CompactionJobQueue`. The new thread that you create reads from the `CompactionJobQueue` and submits jobs until it fills up task slots. If the next job would consume too many slots, it waits for jobs to finish before submitting the next one.
In this way, the Duty can change the state of CompactionJobQueue whenever it thinks that it should be changed and the extra thread can submit jobs and wait for their completion regardless of the duty's scheduled runtime.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org