GH-2269: Improve DLPR Extensibility

Resolves #2269

- add getters for fields used in protected methods
- change more methods to protected

**Cherry-pick to 2.9.x, 2.8.x, 2.7.x**

If it doesn't cherry-pick cleanly, I will back-port.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

@@ -310,7 +310,18 @@ public void setFailIfSendResultIsError(boolean failIfSendResultIsError) {
 	}
 
 	/**
-	 * Set the minumum time to wait for message sending. Default is the producer
+	 * If true, wait for the send result and throw an exception if it fails.
+	 * It will wait for the milliseconds specified in waitForSendResultTimeout for the result.
+	 * @return true to wait.
+	 * @since 2.7.14
+	 * @see #setWaitForSendResultTimeout(Duration)
+	 */
+	protected boolean isFailIfSendResultIsError() {
+		return this.failIfSendResultIsError;
+	}
+
+	/**
+	 * Set the minimum time to wait for message sending. Default is the producer
 	 * configuration {@code delivery.timeout.ms} plus the {@link #setTimeoutBuffer(long)}.
 	 * @param waitForSendResultTimeout the timeout.
 	 * @since 2.7
@@ -322,8 +333,9 @@ public void setWaitForSendResultTimeout(Duration waitForSendResultTimeout) {
 	}
 
 	/**
-	 * Set the number of milliseconds to add to the producer configuration {@code delivery.timeout.ms}
-	 * property to avoid timing out before the Kafka producer. Default 5000.
+	 * Set the number of milliseconds to add to the producer configuration
+	 * {@code delivery.timeout.ms} property to avoid timing out before the Kafka producer.
+	 * Default 5000.
 	 * @param buffer the buffer.
 	 * @since 2.7
 	 * @see #setWaitForSendResultTimeout(Duration)
@@ -332,6 +344,16 @@ public void setTimeoutBuffer(long buffer) {
 		this.timeoutBuffer = buffer;
 	}
 
+	/**
+	 * The number of milliseconds to add to the producer configuration
+	 * {@code delivery.timeout.ms} property to avoid timing out before the Kafka producer.
+	 * @return the buffer.
+	 * @since 2.7.14
+	 */
+	protected long getTimeoutBuffer() {
+		return this.timeoutBuffer;
+	}
+
 	/**
 	 * Set to false to retain previous exception headers as well as headers for the
 	 * current exception. Default is true, which means only the current headers are
@@ -366,6 +388,15 @@ public void setExceptionHeadersCreator(ExceptionHeadersCreator headersCreator) {
 		this.exceptionHeadersCreator = headersCreator;
 	}
 
+	/**
+	 * True if publishing should run in a transaction.
+	 * @return true for transactional.
+	 * @since 2.7.14
+	 */
+	protected boolean isTransactional() {
+		return this.transactional;
+	}
+
 	/**
 	 * Clear the header inclusion bit for the header name.
 	 * @param headers the headers to clear.
@@ -629,7 +660,14 @@ protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations
 		}
 	}
 
-	private void verifySendResult(KafkaOperations<Object, Object> kafkaTemplate,
+	/**
+	 * Wait for the send future to complete.
+	 * @param kafkaTemplate the template used to send the record.
+	 * @param outRecord the record.
+	 * @param sendResult the future.
+	 * @param inRecord the original consumer record.
+	 */
+	protected void verifySendResult(KafkaOperations<Object, Object> kafkaTemplate,
 			ProducerRecord<Object, Object> outRecord,
 			@Nullable ListenableFuture<SendResult<Object, Object>> sendResult, ConsumerRecord<?, ?> inRecord) {
 
@@ -655,7 +693,14 @@ private String pubFailMessage(ProducerRecord<Object, Object> outRecord, Consumer
 				+ outRecord.topic() + "failed for: " + ListenerUtils.recordToString(inRecord, true);
 	}
 
-	private Duration determineSendTimeout(KafkaOperations<?, ?> template) {
+	/**
+	 * Determine the send timeout based on the template's producer factory and
+	 * {@link #setWaitForSendResultTimeout(Duration)}.
+	 * @param template the template.
+	 * @return the timeout.
+	 * @since 2.7.14
+	 */
+	protected Duration determineSendTimeout(KafkaOperations<?, ?> template) {
 		ProducerFactory<? extends Object, ? extends Object> producerFactory = template.getProducerFactory();
 		if (producerFactory != null) { // NOSONAR - will only occur in mock tests
 			Map<String, Object> props = producerFactory.getConfigurationProperties();