Adds cluster level idleConfig setting for supervisor #13311
Conversation
...ce/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfig.java
Outdated
Show resolved
Hide resolved
.../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
Outdated
Show resolved
Hide resolved
.../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
Outdated
Show resolved
Hide resolved
| @@ -46,6 +46,12 @@ | |||
| @JsonProperty | |||
| private int maxStoredExceptionEvents = Math.max(unhealthinessThreshold, healthinessThreshold); | |||
|
|
|||
| @JsonProperty("idleConfig.enabled") | |||
| private boolean idleConfigEnabled = false; | |||
There was a problem hiding this comment.
Maybe instead of flattening the spec, keep an idleConfig itself here and handle it correctly inside the SeekableStreamSupervisor.
There was a problem hiding this comment.
We're passing 2 spec mappings druid.supervisor.idleConfig.enabled, druid.supervisor.idleConfig.inactiveAfterMillis rather than an idleConfig json object, I am not sure if it can be deserialised to idleConfig mapping.
|
Thanks for the review @kfaraz, I have addressed your comments in the latest commit. Please take a look. |
| @@ -804,6 +805,15 @@ public SeekableStreamSupervisor( | |||
| : Math.min(10, this.ioConfig.getTaskCount() * this.ioConfig.getReplicas())); | |||
| } | |||
|
|
|||
| if (spec.getIoConfig().getIdleConfig() != null) { | |||
| idleConfig = spec.getIoConfig().getIdleConfig(); | |||
There was a problem hiding this comment.
It's possible that users have only specified enabled/disabled at the supervisor level, but not the timeout value. Shouldn't we pick up the timeout value defined at the cluster level then?
|
I have addressed your comments in the latest commit @kfaraz. PTAL. |
| @@ -804,6 +805,23 @@ public SeekableStreamSupervisor( | |||
| : Math.min(10, this.ioConfig.getTaskCount() * this.ioConfig.getReplicas())); | |||
| } | |||
|
|
|||
| IdleConfig specIdleConfig = spec.getIoConfig().getIdleConfig(); | |||
| if (specIdleConfig != null) { | |||
| if (specIdleConfig.getInactiveAfterMillis() != null) { | |||
There was a problem hiding this comment.
I feel we might still incorrectly handle the case where user has specified inactiveAfterMillis at the supervisor level but not enabled. In that case, we should just override the inactiveAfterMillis and take the cluster level value for enabled.
But I guess that's a corner case and we can do it in a later PR.
|
@tejaswini-imply , thanks for the fix. A couple of ITs seem to be failing consistently. Could you please check if the failures are genuine? |
|
@kfaraz The transactional kafka index integration test seems to be flaky. It passed in the other run, and the changes have nothing to do with these tests. |
|
Merged since IT failure is unrelated. |
This PR enables cluster admin to define default idle config for all supervisors. The new cluster-level settings and their defaults are:
This configuration can be overridden on individual basis through
ingestionSpec.ioConfig.idleConfig.Key changed/added classes in this PR
SupervisorStateManagerConfigSeekableStreamSupervisorThis PR has: