<p>Keda configuration is as below</p>
<pre><code>apiVersion: keda.sh/v1alpha1
kind: ScaledJob
metadata:
  name: job-message-consumer
spec:
  jobTargetRef:
    parallelism: 1
    template:
      spec:
        containers:
          - name: {{ .Values.JobMessageConsumer.name }}
            image: {{ .Values.image.repository }}
            args:
              - python
              - manage.py
              - jobs_consumer
            imagePullPolicy: {{ .Values.image.pullPolicy }}
            env:
              - name: SSL_CA_LOCATION
                value: {{ .Values.kafka.sslCaDirMount }}/ca.crt
              - name: SSL_CA_DIR_MOUNT
                value: {{ .Values.kafka.sslCaDirMount }}
              - name: SSL_SECRET_NAME
                value: {{ .Values.kafka.sslSecretName }}
              - name: SERVICE_CONSUMER_GROUP_ID
                value: {{ .Values.kafka.CONSUMER_GROUP_ID }}
              - name: KAFKA_SERVICE_JOBS_TOPIC
                value: {{ .Values.job_consumer_service.consumption_topic }}
              {{- include &quot;env&quot; . | nindent 14 }}
            resources:
              limits:
                cpu: &quot;2&quot;
                memory: &quot;4Gi&quot;
              requests:
                cpu: &quot;500m&quot;
                memory: &quot;500Mi&quot;
            volumeMounts:
              - name: user-certs
                mountPath: {{ .Values.kafka.sslUserCertDir }}
              - name: ca-certs
                mountPath: {{ .Values.kafka.sslCaDirMount }}
        volumes:
          - name: user-certs
            secret:
              secretName: {{ .Release.Name }}-kafka-user-certs
          - name: ca-certs
            secret:
              secretName: kafka-cluster-ca-cert
  pollingInterval: 20
  maxReplicaCount: 100
  successfulJobsHistoryLimit: 5
  failedJobsHistoryLimit: 5
  triggers:
    - type: kafka
      metadata:
        bootstrapServers: &quot;kafka-kafka-bootstrap.kafka:9093&quot;
        consumerGroup: &quot;JobConsumer&quot;
        topic: {{ .Values.job_consumer_service.consumption_topic }}
        lagThreshold: &quot;1&quot;
        offsetResetPolicy: latest
      authenticationRef:
        name: {{ .Release.Name }}-keda-trigger-auth-kafka-credential
</code></pre>
<p>The keda-jobs are required to run long time to complete the task hence the ScaledJob. But with the  above ScaledJobs configuration the keda jobs are scaling every 20 sec . I want the keda jobs to scale only based on the messages on the kafka-trigger and not based on the pollinginterval. Can anyone help me on this</p>

Zaks
<p>Please add this parameter to the <code>kafka</code> trigger in the <code>triggers</code> section:</p>
<pre><code>        scaleToZeroOnInvalidOffset: &quot;true&quot;
</code></pre>
<p>If this <code>scaleToZeroOnInvalidOffset</code> parameter, added in versions <code>2.7.0</code> and above, is set to <code>&quot;true&quot;</code>, the consumers for a partition will be scaled to <code>zero</code> when that partition does not have a valid offset [1].</p>
<p>[1] - <a href="https://keda.sh/docs/2.7/scalers/apache-kafka/" rel="nofollow noreferrer">https://keda.sh/docs/2.7/scalers/apache-kafka/</a></p>
