Remove maxInFlight

spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java

@@ -142,16 +142,6 @@
 	 */
 	String concurrency() default "";
 
-	/**
-	 * Override the container factory's {@code maxInFlight} setting for this listener. May
-	 * be a property placeholder or SpEL expression that evaluates to a {@link Number}, in
-	 * which case {@link Number#intValue()} is used to obtain the value.
-	 * <p>
-	 * SpEL {@code #{...}} and property placeholders {@code ${...}} are supported.
-	 * @return the maximum in-flight messages in the reactive client pipeline.
-	 */
-	String maxInFlight() default "";
-
 	/**
 	 * Set to true or false, to override the default setting in the container factory. May
 	 * be a property placeholder or SpEL expression that evaluates to a {@link Boolean} or

spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java

@@ -354,10 +354,6 @@ private void processReactivePulsarListenerAnnotation(MethodReactivePulsarListene
 		if (StringUtils.hasText(concurrency)) {
 			endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency"));
 		}
-		String maxInFlight = reactivePulsarListener.maxInFlight();
-		if (StringUtils.hasText(maxInFlight)) {
-			endpoint.setMaxInFlight(resolveExpressionAsInteger(maxInFlight, "maxInFlight"));
-		}
 		String useKeyOrderedProcessing = reactivePulsarListener.useKeyOrderedProcessing();
 		if (StringUtils.hasText(useKeyOrderedProcessing)) {
 			endpoint.setUseKeyOrderedProcessing(

spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java

@@ -76,8 +76,6 @@ public abstract class AbstractReactivePulsarListenerEndpoint<T>
 
 	private Integer concurrency;
 
-	private Integer maxInFlight;
-
 	private Boolean useKeyOrderedProcessing;
 
 	@Override
@@ -231,15 +229,6 @@ public void setConcurrency(Integer concurrency) {
 		this.concurrency = concurrency;
 	}
 
-	@Override
-	public Integer getMaxInFlight() {
-		return this.maxInFlight;
-	}
-
-	public void setMaxInFlight(Integer maxInFlight) {
-		this.maxInFlight = maxInFlight;
-	}
-
 	@Override
 	public Boolean getUseKeyOrderedProcessing() {
 		return this.useKeyOrderedProcessing;

spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java

@@ -123,13 +123,6 @@ public DefaultReactivePulsarMessageListenerContainer<T> createContainerInstance(
 			properties.setConcurrency(this.containerProperties.getConcurrency());
 		}
 
-		if (endpoint.getMaxInFlight() != null) {
-			properties.setMaxInFlight(endpoint.getMaxInFlight());
-		}
-		else {
-			properties.setMaxInFlight(this.containerProperties.getMaxInFlight());
-		}
-
 		if (endpoint.getUseKeyOrderedProcessing() != null) {
 			properties.setUseKeyOrderedProcessing(endpoint.getUseKeyOrderedProcessing());
 		}

spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java

@@ -32,9 +32,6 @@ public interface ReactivePulsarListenerEndpoint<T> extends ListenerEndpoint<Reac
 
 	boolean isFluxListener();
 
-	@Nullable
-	Integer getMaxInFlight();
-
 	@Nullable
 	Boolean getUseKeyOrderedProcessing();
 

spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java

@@ -85,11 +85,6 @@ public boolean isFluxListener() {
 		return false;
 	}
 
-	@Override
-	public Integer getMaxInFlight() {
-		return null;
-	}
-
 	@Override
 	public Boolean getUseKeyOrderedProcessing() {
 		return null;

spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java

@@ -181,8 +181,7 @@ private ReactiveMessagePipeline startPipeline(ReactivePulsarContainerProperties<
 					.handlingTimeout(containerProperties.getHandlingTimeout());
 			if (containerProperties.getConcurrency() > 0) {
 				ConcurrentOneByOneMessagePipelineBuilder<T> concurrentPipelineBuilder = messagePipelineBuilder
-						.concurrent().concurrency(containerProperties.getConcurrency())
-						.maxInflight(containerProperties.getMaxInFlight());
+						.concurrent().concurrency(containerProperties.getConcurrency());
 				if (containerProperties.isUseKeyOrderedProcessing()) {
 					concurrentPipelineBuilder.useKeyOrderedProcessing();
 				}

spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java

@@ -50,8 +50,6 @@ public class ReactivePulsarContainerProperties<T> {
 
 	private int concurrency = 0;
 
-	private int maxInFlight = 0;
-
 	private boolean useKeyOrderedProcessing = false;
 
 	public ReactivePulsarMessageHandler getMessageHandler() {
@@ -130,14 +128,6 @@ public void setConcurrency(int concurrency) {
 		this.concurrency = concurrency;
 	}
 
-	public int getMaxInFlight() {
-		return this.maxInFlight;
-	}
-
-	public void setMaxInFlight(int maxInFlight) {
-		this.maxInFlight = maxInFlight;
-	}
-
 	public boolean isUseKeyOrderedProcessing() {
 		return this.useKeyOrderedProcessing;
 	}

spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java

@@ -145,7 +145,6 @@ void containerProperties() throws Exception {
 		pulsarContainerProperties.setSubscriptionName(subscriptionName);
 		pulsarContainerProperties.setConcurrency(5);
 		pulsarContainerProperties.setUseKeyOrderedProcessing(true);
-		pulsarContainerProperties.setMaxInFlight(6);
 		pulsarContainerProperties.setHandlingTimeout(Duration.ofMillis(7));
 		DefaultReactivePulsarMessageListenerContainer<String> container = new DefaultReactivePulsarMessageListenerContainer<>(
 				pulsarConsumerFactory, pulsarContainerProperties);
@@ -159,7 +158,7 @@ void containerProperties() throws Exception {
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 
 		assertThat(container).extracting("pipeline", InstanceOfAssertFactories.type(ReactiveMessagePipeline.class))
-				.hasFieldOrPropertyWithValue("concurrency", 5).hasFieldOrPropertyWithValue("maxInflight", 6)
+				.hasFieldOrPropertyWithValue("concurrency", 5)
 				.hasFieldOrPropertyWithValue("handlingTimeout", Duration.ofMillis(7)).extracting("groupingFunction")
 				.isInstanceOf(DefaultMessageGroupingFunction.class);