KAFKA-15429: reset transactionInFlight on StreamsProducer close (#14326)

Resets the value of transactionInFlight to false when closing the
StreamsProducer. This ensures we don't try to commit against a
closed producer

Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java

@@ -192,12 +192,11 @@ public void resetProducer() {
 
         oldProducerTotalBlockedTime += totalBlockedTime(producer);
         final long start = time.nanoseconds();
-        producer.close();
+        close();
         final long closeTime = time.nanoseconds() - start;
         oldProducerTotalBlockedTime += closeTime;
 
         producer = clientSupplier.getProducer(eosV2ProducerConfigs);
-        transactionInitialized = false;
     }
 
     private double getMetricValue(final Map<MetricName, ? extends Metric> metrics,
@@ -371,6 +370,8 @@ void flush() {
 
     void close() {
         producer.close();
+        transactionInFlight = false;
+        transactionInitialized = false;
     }
 
     // for testing only

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java

@@ -207,6 +207,34 @@ public void before() {
 
     // functional tests
 
+    @Test
+    public void shouldResetTransactionInFlightOnClose() {
+        // given:
+        eosBetaStreamsProducer.send(
+            new ProducerRecord<>("topic", new byte[1]), (metadata, error) -> { });
+        assertThat(eosBetaStreamsProducer.transactionInFlight(), is(true));
+
+        // when:
+        eosBetaStreamsProducer.close();
+
+        // then:
+        assertThat(eosBetaStreamsProducer.transactionInFlight(), is(false));
+    }
+
+    @Test
+    public void shouldResetTransactionInFlightOnReset() {
+        // given:
+        eosBetaStreamsProducer.send(
+            new ProducerRecord<>("topic", new byte[1]), (metadata, error) -> { });
+        assertThat(eosBetaStreamsProducer.transactionInFlight(), is(true));
+
+        // when:
+        eosBetaStreamsProducer.resetProducer();
+
+        // then:
+        assertThat(eosBetaStreamsProducer.transactionInFlight(), is(false));
+    }
+
     @Test
     public void shouldCreateProducer() {
         assertThat(mockClientSupplier.producers.size(), is(1));