@Test public void shouldAbortTransactionOnDirtyClosedIfEosEnabled() throws Exception { final MockProducer producer = new MockProducer(); task = new StreamTask(taskId00, applicationId, partitions, topology, consumer, changelogReader, eosConfig, streamsMetrics, stateDirectory, null, time, producer); task.close(false); task = null; assertTrue(producer.transactionAborted()); }
@Override public void close(boolean clean) { log.debug("{} Closing", logPrefix); RuntimeException firstException = null; try { suspend(clean); } catch (final RuntimeException e) { clean = false; firstException = e; log.error("{} Could not close task: ", logPrefix, e); } try { closeStateManager(clean); } catch (final RuntimeException e) { clean = false; if (firstException == null) { firstException = e; } log.error("{} Could not close state manager: ", logPrefix, e); } try { partitionGroup.close(); metrics.removeAllSensors(); } finally { if (eosEnabled) { if (!clean) { try { producer.abortTransaction(); transactionInFlight = false; } catch (final ProducerFencedException e) { } } try { recordCollector.close(); } catch (final Throwable e) { log.error("{} Failed to close producer: ", logPrefix, e); } } } if (firstException != null) { throw firstException; } }
StreamTask extends AbstractTask implements Punctuator { @Override public void close(boolean clean) { log.debug("{} Closing", logPrefix); RuntimeException firstException = null; try { suspend(clean); } catch (final RuntimeException e) { clean = false; firstException = e; log.error("{} Could not close task: ", logPrefix, e); } try { closeStateManager(clean); } catch (final RuntimeException e) { clean = false; if (firstException == null) { firstException = e; } log.error("{} Could not close state manager: ", logPrefix, e); } try { partitionGroup.close(); metrics.removeAllSensors(); } finally { if (eosEnabled) { if (!clean) { try { producer.abortTransaction(); transactionInFlight = false; } catch (final ProducerFencedException e) { } } try { recordCollector.close(); } catch (final Throwable e) { log.error("{} Failed to close producer: ", logPrefix, e); } } } if (firstException != null) { throw firstException; } } }
StreamTask extends AbstractTask implements Punctuator { @Override public void close(boolean clean) { log.debug("{} Closing", logPrefix); RuntimeException firstException = null; try { suspend(clean); } catch (final RuntimeException e) { clean = false; firstException = e; log.error("{} Could not close task: ", logPrefix, e); } try { closeStateManager(clean); } catch (final RuntimeException e) { clean = false; if (firstException == null) { firstException = e; } log.error("{} Could not close state manager: ", logPrefix, e); } try { partitionGroup.close(); metrics.removeAllSensors(); } finally { if (eosEnabled) { if (!clean) { try { producer.abortTransaction(); transactionInFlight = false; } catch (final ProducerFencedException e) { } } try { recordCollector.close(); } catch (final Throwable e) { log.error("{} Failed to close producer: ", logPrefix, e); } } } if (firstException != null) { throw firstException; } } StreamTask(final TaskId id,
                      final String applicationId,
                      final Collection<TopicPartition> partitions,
                      final ProcessorTopology topology,
                      final Consumer<byte[], byte[]> consumer,
                      final ChangelogReader changelogReader,
                      final StreamsConfig config,
                      final StreamsMetrics metrics,
                      final StateDirectory stateDirectory,
                      final ThreadCache cache,
                      final Time time,
                      final Producer<byte[], byte[]> producer); }
StreamTask extends AbstractTask implements Punctuator { @Override public void close(boolean clean) { log.debug("{} Closing", logPrefix); RuntimeException firstException = null; try { suspend(clean); } catch (final RuntimeException e) { clean = false; firstException = e; log.error("{} Could not close task: ", logPrefix, e); } try { closeStateManager(clean); } catch (final RuntimeException e) { clean = false; if (firstException == null) { firstException = e; } log.error("{} Could not close state manager: ", logPrefix, e); } try { partitionGroup.close(); metrics.removeAllSensors(); } finally { if (eosEnabled) { if (!clean) { try { producer.abortTransaction(); transactionInFlight = false; } catch (final ProducerFencedException e) { } } try { recordCollector.close(); } catch (final Throwable e) { log.error("{} Failed to close producer: ", logPrefix, e); } } } if (firstException != null) { throw firstException; } } StreamTask(final TaskId id,
                      final String applicationId,
                      final Collection<TopicPartition> partitions,
                      final ProcessorTopology topology,
                      final Consumer<byte[], byte[]> consumer,
                      final ChangelogReader changelogReader,
                      final StreamsConfig config,
                      final StreamsMetrics metrics,
                      final StateDirectory stateDirectory,
                      final ThreadCache cache,
                      final Time time,
                      final Producer<byte[], byte[]> producer); @Override void resume(); @SuppressWarnings("unchecked") boolean process(); @Override void punctuate(final ProcessorNode node, final long timestamp); @Override void commit(); @Override void suspend(); @Override void close(boolean clean); @SuppressWarnings("unchecked") int addRecords(final TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records); void schedule(final long interval); }
StreamTask extends AbstractTask implements Punctuator { @Override public void close(boolean clean) { log.debug("{} Closing", logPrefix); RuntimeException firstException = null; try { suspend(clean); } catch (final RuntimeException e) { clean = false; firstException = e; log.error("{} Could not close task: ", logPrefix, e); } try { closeStateManager(clean); } catch (final RuntimeException e) { clean = false; if (firstException == null) { firstException = e; } log.error("{} Could not close state manager: ", logPrefix, e); } try { partitionGroup.close(); metrics.removeAllSensors(); } finally { if (eosEnabled) { if (!clean) { try { producer.abortTransaction(); transactionInFlight = false; } catch (final ProducerFencedException e) { } } try { recordCollector.close(); } catch (final Throwable e) { log.error("{} Failed to close producer: ", logPrefix, e); } } } if (firstException != null) { throw firstException; } } StreamTask(final TaskId id,
                      final String applicationId,
                      final Collection<TopicPartition> partitions,
                      final ProcessorTopology topology,
                      final Consumer<byte[], byte[]> consumer,
                      final ChangelogReader changelogReader,
                      final StreamsConfig config,
                      final StreamsMetrics metrics,
                      final StateDirectory stateDirectory,
                      final ThreadCache cache,
                      final Time time,
                      final Producer<byte[], byte[]> producer); @Override void resume(); @SuppressWarnings("unchecked") boolean process(); @Override void punctuate(final ProcessorNode node, final long timestamp); @Override void commit(); @Override void suspend(); @Override void close(boolean clean); @SuppressWarnings("unchecked") int addRecords(final TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records); void schedule(final long interval);  }