POC for stream binder ignoring immediate stop flag

Requirements

Ensure there's unauthenticated kafka broker running at localhost:9092 such as

docker run \
    -d \
    --rm \
    --hostname localhost \
    -p 9092:9092 \
    -e KAFKA_CFG_NODE_ID=0 \
    -e KAFKA_CFG_PROCESS_ROLES=controller,broker \
    -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
    -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
    -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@localhost:9093 \
    -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER \
    docker.io/bitnami/kafka:4.0

Running

env "spring.cloud.stream.kafka.binder.brokers"=localhost:9092 ./mvnw spring-boot:run

Expected output

ContainerProperties [
topics=[foo]
pollTimeout=5000
consumerRebalanceListener=org.springframework.kafka.listener.AbstractMessageListenerContainer$1@e2b3026
syncCommits=true
authExceptionRetryInterval=null
commitRetries=3
fixTxOffsetsfalse
ackMode=BATCH
ackCount=1
ackTime=5000
consumerStartTimeout=PT30S
messageListener=null
shutdownTimeout=10000
idleEventInterval=30000
idlePartitionEventInterval=not enabled
monitorInterval=30
noPollThreshold=3.0
pauseImmediate=false
pollTimeoutWhilePaused=PT0.1S
subBatchPerPartition=null
assignmentCommitOption=LATEST_ONLY_NO_TX
deliveryAttemptHeader=false
batchRecoverAfterRollback=false
eosMode=V2
transactionDefinition=null
stopContainerWhenFenced=false
stopImmediate=true # <--
asyncAcks=false
logContainerConfig=false
missingTopicsFatal=false
idleBeforeDataMultiplier=5.0
idleBetweenPolls=0
micrometerEnabled=true
observationEnabled=false
observationRegistry=io.micrometer.observation.NoopObservationRegistry@14874a5d
restartAfterAuthExceptions=false
]

Actual

ContainerProperties [
topics=[foo]
pollTimeout=5000
consumerRebalanceListener=org.springframework.kafka.listener.AbstractMessageListenerContainer$1@e2b3026
syncCommits=true
authExceptionRetryInterval=null
commitRetries=3
fixTxOffsetsfalse
ackMode=BATCH
ackCount=1
ackTime=5000
consumerStartTimeout=PT30S
messageListener=null
shutdownTimeout=10000
idleEventInterval=30000
idlePartitionEventInterval=not enabled
monitorInterval=30
noPollThreshold=3.0
pauseImmediate=false
pollTimeoutWhilePaused=PT0.1S
subBatchPerPartition=null
assignmentCommitOption=LATEST_ONLY_NO_TX
deliveryAttemptHeader=false
batchRecoverAfterRollback=false
eosMode=V2
transactionDefinition=null
stopContainerWhenFenced=false
stopImmediate=false # <-- should be true
asyncAcks=false
logContainerConfig=false
missingTopicsFatal=false
idleBeforeDataMultiplier=5.0
idleBetweenPolls=0
micrometerEnabled=true
observationEnabled=false
observationRegistry=io.micrometer.observation.NoopObservationRegistry@14874a5d
restartAfterAuthExceptions=false
]