如何配置KEDA ScaledJob实现单Pod处理单条Service Bus消息后终止

我在Azure Kubernetes Service(AKS)平台使用KEDA ScaledJob运行长时任务，通过Azure Service Bus队列触发器自动触发Job。目前向Service Bus添加消息时，KEDA会自动创建节点/Pod，但所有消息被部分Pod处理，需要实现每个扩容的Pod处理单条消息后立即终止。

当前YAML配置 #

apiVersion: keda.sh/v1alpha1
kind: ScaledJob
metadata:
  name: {{ .Chart.Name }}
spec:
  jobTargetRef:
    backoffLimit: 4
    parallelism: 1
    completions: 1
    activeDeadlineSeconds: 300
    template:
      spec:
        imagePullSecrets:
          - name: {{ .Values.image.imagePullSecrets }}
        terminationGracePeriodSeconds: 30
        dnsPolicy: ClusterFirst
        volumes:
          - name: azure
            azureFile:
              shareName: sharenameone
              secretName: secret-sharenameone
              readOnly: true
          - name: one-storage
            emptyDir: {}
          - name: "logging-volume-file"
            persistentVolumeClaim:
              claimName: "azure-file-logging"        
        initContainers:
          - name: test-java-init
            image: {{ .Values.global.imageRegistryURI }}/{{ .Values.image.javaInitImage.name}}:{{ .Values.image.javaInitImage.tag }}
            imagePullPolicy: {{ .Values.image.pullPolicy }}
            securityContext:
              readOnlyRootFilesystem: true
            resources:
              requests:
                cpu: 100m
                memory: 300Mi
              limits:
                cpu: 200m
                memory: 400Mi
            volumeMounts:
              - name: azure
                mountPath: /mnt/azure
              - name: one-storage
                mountPath: /certs
        containers:
          - name: {{ .Chart.Name }}
            image: {{ .Values.global.imageRegistryURI }}/tests/{{ .Chart.Name }}:{{ .Values.version }}
            imagePullPolicy: {{ .Values.image.pullPolicy }}
            env:
              {{- include "chart.envVars" . | nindent 14 }}
              - name: JAVA_OPTS
                value: >-
                    {{ .Values.application.javaOpts }}          
              - name: application_name
                value: "test_application"
              - name: queueName
                value: "test-queue-name"
              - name: servicebusconnstrenv
                valueFrom: 
                  secretKeyRef:
                    name: secrets-service-bus
                    key: service_bus_conn_str
            volumeMounts:
              - name: cert-storage
                mountPath: /certs
              - name: "logging-volume-azure-file"
                mountPath: "/mnt/logging"
            resources:
              {{- toYaml .Values.resources | nindent 14 }}                   
  pollingInterval: 30
  maxReplicaCount: 5
  successfulJobsHistoryLimit: 5
  failedJobsHistoryLimit: 20
  triggers:
  - type: azure-servicebus
    metadata:
      queueName: "test-queue-name"
      connectionFromEnv: servicebusconnstrenv
      messageCount: "1"

当前Azure Function代码 #

@FunctionName("TestServiceBusTrigger")
public void TestServiceBusTriggerHandler(
    @ServiceBusQueueTrigger(
            name = "msg",
            queueName = "%TEST_QUEUE_NAME%",
            connection = "ServiceBusConnectionString")
        final String inputMessage,
    final ExecutionContext context) {

  final java.util.logging.Logger contextLogger = context.getLogger();
  System.setProperty("javax.net.ssl.trustStore", "/certs/cacerts");
 
  try {
    // 消息处理逻辑
  } catch (Exception e) {
    // 异常处理
  }
}

要实现单Pod单消息处理后终止，需从KEDA触发器配置和Function运行模式两方面调整：

1. 优化KEDA ScaledJob触发器配置 #

当前messageCount: "1"已设置单Job对应1条消息，但需添加锁配置避免消息被重复拉取：

triggers:
- type: azure-servicebus
  metadata:
    queueName: "test-queue-name"
    connectionFromEnv: servicebusconnstrenv
    messageCount: "1"
    # 消息锁定时长，需大于单条消息处理时间
    lockDuration: "5m"
    # 最大锁续期时长，防止处理超时导致消息重新分发
    maxLockRenewalDuration: "5m"

2. 修改Azure Function为单次执行模式 #

默认Function是持续监听的宿主模式，会处理多条消息，需改成处理完单条消息后主动退出：

第一步：添加环境变量 #

在容器env部分新增以下配置：

env:
  {{- include "chart.envVars" . | nindent 14 }}
  # 原有环境变量保留...
  - name: FUNCTIONS_WORKER_RUNTIME
    value: java
  - name: FUNCTIONS_WORKER_PROCESS_COUNT
    value: "1"
  - name: AzureWebJobsDisableHomepage
    value: "true"

第二步：修改代码主动退出进程 #

在消息处理完成后调用System.exit()终止进程：

@FunctionName("TestServiceBusTrigger")
public void TestServiceBusTriggerHandler(
    @ServiceBusQueueTrigger(
            name = "msg",
            queueName = "%TEST_QUEUE_NAME%",
            connection = "ServiceBusConnectionString")
        final String inputMessage,
    final ExecutionContext context) {

  final java.util.logging.Logger contextLogger = context.getLogger();
  System.setProperty("javax.net.ssl.trustStore", "/certs/cacerts");
 
  try {
    // 消息处理逻辑
    contextLogger.info("单条消息处理完成: " + inputMessage);
    // 处理完成后主动退出进程
    System.exit(0);
  } catch (Exception e) {
    contextLogger.severe("消息处理失败: " + e.getMessage());
    // 失败也退出，让KEDA重新触发Job处理该消息
    System.exit(1);
  }
}

3. 调整Pod终止配置 #

缩短终止宽限期，让Pod处理完成后快速退出：

terminationGracePeriodSeconds: 10

内容的提问来源于stack exchange，提问作者Developer208