GH-1618: Remove Duplicate Parameter

Resolves #1618

Incorrectly added the child container as a parameter for the `ConsumerStoppedEvent`.
The child container is already passed in as the `source`.

**cherry-pick to 2.5.x**

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStoppedEvent.java

@@ -16,8 +16,6 @@
 
 package org.springframework.kafka.event;
 
-import org.springframework.lang.Nullable;
-
 /**
  * An event published when a consumer is stopped. While it is best practice to use
  * stateless listeners, you can consume this event to clean up any thread-based resources
@@ -55,8 +53,6 @@ public enum Reason {
 
 	private final Reason reason;
 
-	private final Object container;
-
 	/**
 	 * Construct an instance with the provided source.
 	 * @param source the container.
@@ -74,22 +70,18 @@ public ConsumerStoppedEvent(Object source) {
 	 */
 	@Deprecated
 	public ConsumerStoppedEvent(Object source, Object container) {
-		this(source, container, null, Reason.NORMAL);
+		this(source, container, Reason.NORMAL);
 	}
 
 	/**
 	 * Construct an instance with the provided source and container.
 	 * @param source the container instance that generated the event.
 	 * @param container the container or the parent container if the container is a child.
-	 * @param childContainer the child container, or null.
 	 * @param reason the reason.
 	 * @since 2.5.8
 	 */
-	public ConsumerStoppedEvent(Object source, Object container, @Nullable Object childContainer,
-			Reason reason) {
-
+	public ConsumerStoppedEvent(Object source, Object container, Reason reason) {
 		super(source, container);
-		this.container = childContainer;
 		this.reason = reason;
 	}
 
@@ -102,22 +94,9 @@ public Reason getReason() {
 		return this.reason;
 	}
 
-	/**
-	 * Return the container that the Consumer belonged to.
-	 * @param <T> the container type.
-	 * @return the container.
-	 * @since 2.5.8
-	 */
-	@SuppressWarnings("unchecked")
-	public <T> T getContainer() {
-		return this.container == null ? (T) getSource() : (T) this.container;
-	}
-
 	@Override
 	public String toString() {
-		return "ConsumerStoppedEvent [source=" + getSource()
-				+ (this.container == null ? "" : (", container=" + this.container))
-				+ ", reason=" + this.reason + "]";
+		return "ConsumerStoppedEvent [source=" + getSource() + ", reason=" + this.reason + "]";
 	}
 
 }

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

@@ -397,7 +397,6 @@ private void publishConsumerStoppingEvent(Consumer<?, ?> consumer) {
 	private void publishConsumerStoppedEvent(@Nullable Throwable throwable) {
 		if (getApplicationEventPublisher() != null) {
 			getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this, this.thisOrParentContainer,
-					this.thisOrParentContainer.equals(this) ? null : this,
 					throwable instanceof Error
 						? Reason.ERROR
 						: throwable instanceof StopAfterFenceException

spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java

@@ -333,8 +333,7 @@ public void onMessage(Object data) {
 			assertThat(stopEventLatch.await(10, TimeUnit.SECONDS)).isTrue();
 			assertThat(stopEvent.get().getReason()).isEqualTo(Reason.NORMAL);
 		}
-		MessageListenerContainer stoppedContainer = stopEvent.get().getContainer();
-		assertThat(stoppedContainer).isSameAs(container);
+		assertThat(stopEvent.get().getSource()).isSameAs(container);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })

src/reference/asciidoc/kafka.adoc

@@ -2533,6 +2533,24 @@ The `ConsumerStartingEvent`, `ConsumerStartingEvent`, `ConsumerFailedToStartEven
 All containers (whether a child or a parent) publish `ContainerStoppedEvent`.
 For a parent container, the source and container properties are identical.
 
+In addition, the `ConsumerStoppedEvent` has the following additional property:
+
+* `reason`
+** `NORMAL` - the consumer stopped normally (container was stopped).
+** `ERROR` - a `java.lang.Error` was thrown.
+** `FENCED` - the transactional producer was fenced and the `stopContainerWhenFenced` container property is `true`.
+
+You can use this event to restart the container after such a condition:
+
+====
+[source, java]
+----
+if (event.getReason.equals(Reason.FENCED)) {
+    event.getSource(MessageListenerContainer.class).start();
+}
+----
+====
+
 [[idle-containers]]
 ===== Detecting Idle and Non-Responsive Consumers
 
@@ -4873,6 +4891,8 @@ IMPORTANT: With current `kafka-clients`, the container cannot detect whether a `
 Because, in most cases, it is caused by a rebalance, the container does not call the `AfterRollbackProcessor` (because it's not appropriate to seek the partitions because we no longer are assigned them).
 If you ensure the timeout is large enough to process each transaction and periodically perform an "empty" transaction (e.g. via a `ListenerContainerIdleEvent`) you can avoid fencing due to timeout and expiry.
 Or, you can set the `stopContainerWhenFenced` container property to `true` and the container will stop, avoiding the loss of records.
+You can consume a `ConsumerStoppedEvent` and check the `Reason` property for `FENCED` to detect this condition.
+Since the event also has a reference to the container, you can restart the container using this event.
 
 [[delivery-header]]
 ===== Delivery Attempts Header