Samzaメッセージを送信できませんでした。例外

Question

私はaws emrインスタンスでsamzaを使用していますが、私はいつも次のような例外があります。誰でも助けてくれますか？：

org.apache.samza.SamzaException：メッセージの送信に失敗しました。例外： java.lang.IllegalStateException：プロデューサが閉じられた後に送信できません。 at org.apache.samza.system.kafka.KafkaSystemProducer $$ anonfun $ send $ 5.apply（KafkaSystemProducer.scala：120） at org.apache.samza.system.kafka.KafkaSystemProducer $$ anonfun $ send $ 5.apply（ KafkaSystemProducer.scala：111） org.apache.samza.util.ExponentialSleepStrategy.run（ExponentialSleepStrategy.scalaで：で 86）：81） org.apache.samza.system.kafka.KafkaSystemProducer.sendで（KafkaSystemProducer.scala org.apache.samza.system.SystemProducers.send（SystemProducers.scala：87） at org.apache.samza.task.TaskInstanceCollector.send（TaskInstanceCollector.scala：61） at org.apache.samza.storage.kv。 LoggedStore.putAll（LoggedStore.scala：72） at org.apache.samza.storage.kv.SerializedKeyValueStore.putAll（SerializedKeyValueStore.scala：57）（NullSafeKeyValueStore.scala：69） at org.apache。 samza.storage.kv.KeyValueStorageEngine.flush（KeyValueStorageEngine.scala：113） at org.apache.samza.storage.kv.KeyValueStorageEngine.close（KeyValueStorageEngine.scala：125） at org.apache.samza.storage.kv。 KeyValueStorageEngine.stop（KeyValueStorageEngine.scala：119） at org.apache.samza.storage.TaskStorageManager $$ anonfun $ stop $ 2.apply（TaskStorageManager.scala：147） at org.apache.samza.storage.TaskStorageManager $$ anonfun $ stop $ 2.apply（TaskStorageManager.scala：147） at scala.collection.Iterator $ class.foreach（Iterator.scala：727） at scala.collection.Abstr actIterator.foreach（Iterator.scala：1157） scala.collection.MapLike $ DefaultValuesIterable.foreach（MapLike.scala：206）で org.apache.samza.storage.TaskStorageManager.stop（TaskStorageManager.scala：147）で でorg.apache.samza.container.TaskInstance.shutdownStores（TaskInstance.scala：185） at org.apache.samza.container.SamzaContainer $$ anonfun $ shutdownStores $ 2.apply（SamzaContainer.scala：650） at org.apache。 $ this.AbstractIterator.foreach（Iterator.Scala）：$ this.basic.container.SamzaContainer $$ anonfun $ shutdownStores $ 2.apply（SamzaContainer.scala：650） at scala.collection.Iterator $ class.foreach（Iterator.scala：727） at scala.collection.AbstractIterator.foreach scala：1157） at scala.collection.MapLike $ DefaultValuesIterable.foreach（MapLike.scala：206） at org.ap （SamzaContainer.scala：650） at org.apache.samza.container.SamzaContainer.run （SamzaContainer.scala：93） org.apache.samza.container.SamzaContainer $ .mainで（SamzaContainer.scala：67） org.apache.samza.container.SamzaContainer.mainで（SamzaContainer.scala）

Answer 1

私は間違ったファイルで実際の例外を逃しました。