SO-61650223

application.yml

spring:
  cloud:
    stream:
      bindings:
        source:
          content-type: text/plain
          group: so61650223
      kafka:
        bindings:
          source:
            consumer:
              poll-timeout: 10000

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.2.6.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.example</groupId>
	<artifactId>so61650223</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>so61650223</name>
	<description>Demo project for Spring Boot</description>

	<properties>
		<java.version>1.8</java.version>
		<spring-cloud.version>Hoxton.SR4</spring-cloud.version>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
			<exclusions>
				<exclusion>
					<groupId>org.junit.vintage</groupId>
					<artifactId>junit-vintage-engine</artifactId>
				</exclusion>
			</exclusions>
		</dependency>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream-test-support</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-actuator</artifactId>
		</dependency>
	</dependencies>

	<dependencyManagement>
		<dependencies>
			<dependency>
				<groupId>org.springframework.cloud</groupId>
				<artifactId>spring-cloud-dependencies</artifactId>
				<version>${spring-cloud.version}</version>
				<type>pom</type>
				<scope>import</scope>
			</dependency>
		</dependencies>
	</dependencyManagement>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>

So61650223Application.java

/*
 * Copyright 2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.example.demo;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.binder.PollableMessageSource;
import org.springframework.cloud.stream.config.MessageSourceCustomizer;
import org.springframework.cloud.stream.endpoint.BindingsEndpoint;
import org.springframework.cloud.stream.endpoint.BindingsEndpoint.State;
import org.springframework.context.annotation.Bean;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.acks.AcknowledgmentCallback;
import org.springframework.integration.acks.AcknowledgmentCallback.Status;
import org.springframework.integration.kafka.inbound.KafkaMessageSource;
import org.springframework.stereotype.Component;

@SpringBootApplication
@EnableBinding(Poller.class)
public class So61650223Application {

	public static void main(String[] args) {
		SpringApplication.run(So61650223Application.class, args);
	}

	@Bean
	MessageSourceCustomizer<KafkaMessageSource<byte[], byte[]>> customizer() {
		return (source, dest, group) -> {
			source.getConsumerProperties().getKafkaConsumerProperties()
					.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
		};
	}

	@Bean
	TaskExecutor executor() {
		return new SimpleAsyncTaskExecutor();
	}


	@Bean
	public ApplicationRunner runner(Worker worker) {
		return args -> worker.run();
	}

}

interface Poller {

	@Input
	PollableMessageSource source();

}

@Component
class Worker implements Runnable, DisposableBean {

	private static final Logger LOG = LoggerFactory.getLogger(Worker.class);

	private final PollableMessageSource source;

	private final TaskExecutor executor;

	private final BindingsEndpoint endpoint;

	private volatile boolean okToRun = true;

	private volatile boolean jobDone = true;

	private volatile boolean success;

	private volatile CountDownLatch latch;

	public Worker(PollableMessageSource source, TaskExecutor executor, BindingsEndpoint endpoint) {
		this.source = source;
		this.executor = executor;
		this.endpoint = endpoint;
	}

	@Override
	public void run() {

		AtomicReference<AcknowledgmentCallback> callback = new AtomicReference<>();
		try {
			while (okToRun) {

				while (this.okToRun) {
					LOG.info("Polling for work");
					if (this.source.poll(msg -> {
							LOG.info("Work found");
							callback.set(StaticMessageHeaderAccessor.getAcknowledgmentCallback(msg));
							callback.get().noAutoAck();
							this.jobDone = false;
							this.success = true;
							this.latch = new CountDownLatch(1);
							this.executor.execute(() -> runLongJob((String) msg.getPayload()));
						}, new ParameterizedTypeReference<String>() {})) {
						break;
					}
				}
				this.endpoint.changeState("input", State.PAUSED);

				while (this.okToRun && !this.jobDone) {
					LOG.info("Polling while paused");
					this.source.poll(msg -> { });
					this.latch.await(10, TimeUnit.SECONDS);
				}

				if (this.success) {
					LOG.info("Accepted");
					callback.get().acknowledge(Status.ACCEPT);
				}
				else {
					LOG.info("Requeued");
					callback.get().acknowledge(Status.REQUEUE);
				}

				this.endpoint.changeState("input", State.RESUMED);
			}
		}
		catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}

	}

	private void runLongJob(String work) {
		LOG.info("Running long job: " + work);
		try {
			Thread.sleep(60_000);
		}
		catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
		LOG.info("Job done");
		this.jobDone = true;
//		this.success = !"fail".equals(work); // test requeue
		this.latch.countDown();
	}

	@Override
	public void destroy() {
		this.okToRun = false;
	}

}