Parallel consumer pods monitoring by Kubernetes

[jikeshdpatel]

Hi,
We are using Kubernetes to orchestrate parallel consumer pods. I know it's not perfect for Kafka consumer auto scaling but at least we want to get pods auto created if PC is leaving the consumer group with below exception. Currently the issue I am facing is - with this exception the app is not shutdown and the health endpoints are still alive so Kubernetes does not replace the pod which is not registered or subscribed to any of the partitions.

Error from poll control thread, will attempt controlled shutdown, then rethrow. Error: Timeout waiting for commit response PT30S to request ConsumerOffsetCommitter.CommitRequest(id=xxxxxx, requestedAtMs=xxxxx)

Would really appreciate any help on this!

Thanks in advance.

[jikeshdpatel]

Implement custom HealthIndicator and return 503 on ParallelEoSStreamProcessor.isClosedOrFailed.
For example :

import com.jpatel.test.confluent.kafka.ParallelConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.stereotype.Component;

@Component("pcHealthIndicator")
public class PCHealthIndicator implements HealthIndicator {

    @Autowired
    private ParallelConsumer parallelConsumer;

    @Override
    public Health health() {
        if (parallelConsumer.getParallelStreamProcessor().isClosedOrFailed()) {
            logger.warn("PCHealthIndicator :: health reports parallel consumer is closed or failed" +
                    " nr of assigned partitions {} , failure root cause {} , "
                    , parallelConsumer.getParallelStreamProcessor().getNumberOfAssignedPartitions()
                    , parallelConsumer.getParallelStreamProcessor().getFailureCause()
                    , parallelConsumer.getParallelStreamProcessor().getMyId()
            );
            return Health.down().build();
        } else {
            logger.debug("PCHealthIndicator :: health reports parallel consumer is still up" +
                            " nr of assigned partitions {} , failure root cause {} , ids {}"
                    , parallelConsumer.getParallelStreamProcessor().getNumberOfAssignedPartitions()
                    , parallelConsumer.getParallelStreamProcessor().getFailureCause()
                    , parallelConsumer.getParallelStreamProcessor().getMyId()
            );
            return Health.up().build();
        }
    }
}

Expose to spring health management

management:
  server:
    port: 8081
  endpoints:
    web:
      exposure:
        include: health,liveness,readiness
  endpoint:
    health:
      show-details: always
      probes:
        enabled: true
    group:
      readiness:
        include: pcHealthIndicator
      liveness:
        include: pcHealthIndicator

Change kubernetes liveliness endpoint to pcHealthIndicator in terraform module

  readiness_probe_path   = "/actuator/health/readiness"
  liveness_probe_path    = "/actuator/health/pcHealthIndicator"

[astubbs]

Welcome to the project! Sorry I just saw this, this morning. Thanks for pinging again. What version are you using? And can you include more logs before the error?

[jikeshdpatel]

Thank you @astubbs for taking a look here! I have been using this library in prod and able to get great throughput without increasing partitions. Only limit now is Kafka broker throughput and I/O processing in the consumer. Really appreciate it!

So far, I noticed this and issue#351 issues otherwise it's working great!

Related to this discussion, I am running PC from Spring boot app. PC unregisters from Kafka broker when it encounters below error due to any kind of reason. I am not sure if we have any PC health indicator endpoint so I have created a custom health HealthIndicator and it is now working fine. It would be extremely helpful if such health endpoint is implemented while integrating with Spring.

Error from poll control thread, will attempt controlled shutdown, then rethrow. Error:

I have a test case to reproduce this particular issue. I can contribute it if you think that's worth it.

[astubbs]

I have a test case to reproduce this particular issue. I can contribute it if you think that's worth it.

Yes, definitely! Please do so.

Also - there's a health monitoring issue request - please upvote on it / comment what you'd like to see:

• Health-checks #71

[RavindraDobariya]

Hi @astubbs i am getting same error with spring boot while rebalancing is happening. below are the steps to reproduce the scenario.

using PERIODIC_CONSUMER_SYNC mode with 1 sec interval and CooperativeStickyAssignor.

1. run app on 1 pod lets say pod1. all partitions will be assigned to pod1.
2. scale to 2 pods. now let's say new pod is pod2.
3. once pod2 is un and running pod1 getting below error.

org.apache.kafka.common.errors.RebalanceInProgressException: Offset commit cannot be completed since the consumer is undergoing a rebalance for auto partition assignment. You can try completing the rebalance by calling poll() and then retry the operation.

and after that
io.confluent.parallelconsumer.internal.InternalRuntimeException: Timeout waiting for commit response PT30S to request ConsumerOffsetCommitter.CommitRequest(id=1f96355e-bdc7-4f39-9102-db6a408245d5, requestedAtMs=1708296847469)

now pod1 has no cosumer and all parititions will be assigned to pod2. so i am not able to scale application. can you please help on solution.