Samza Не удалось отправить сообщение. Исключение

Question

Я использую samza в экземплярах aws emr, и у меня всегда есть исключение, например следующее: может ли кто-нибудь мне помочь ?:

org.apache.samza.SamzaException: Не удалось отправить сообщение. Исключение: java.lang.IllegalStateException: Не удается отправить после закрытия производителя. at org.apache.samza.system.kafka.KafkaSystemProducer $$ anonfun $ отправить $ 5.apply (KafkaSystemProducer.scala: 120) на org.apache.samza.system.kafka.KafkaSystemProducer $$ anonfun $ отправить $ 5.apply (KafkaSystemProducer.scala: 111) на org.apache.samza.util.ExponentialSleepStrategy.run (ExponentialSleepStrategy.scala: 81) на org.apache.samza.system.kafka.KafkaSystemProducer.send (KafkaSystemProducer.scala: 86) на 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) at org.apache.samza.storage.kv.CachedStore.flush (CachedStore.scala: 166) at org.apache.samza.storage.kv.NullSafeKeyValueStore.flush (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) at scala.collection.MapLike $ DefaultValuesIterable.foreach (MapLike.scala: 206) at org.apache.samza.storage.TaskStorageManager.stop (TaskStorageManager.scala: 147) at 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. samza.container.SamzaContainer $$ anonfun $ shutdownStores $ 2.apply (SamzaContainer.scala: 650) at scala.collection.Iterator $ class.foreach (Iterator.scala: 727) at scala.collection.AbstractIterator.foreach (Iterator. scala: 1157) at scala.collection.MapLike $ DefaultValuesIterable.foreach (MapLike.scala: 206) at org.ap ache.samza.container.SamzaContainer.shutdownStores (SamzaContainer.scala: 650) at org.apache.samza.container.SamzaContainer.run (SamzaContainer.scala: 560) at org.apache.samza.container.SamzaContainer $ .safeMain (SamzaContainer.scala: 93) в org.apache.samza.container.SamzaContainer $ .main (SamzaContainer.scala: 67) на org.apache.samza.container.SamzaContainer.main (SamzaContainer.scala)

Answer 1

Я пропустил настоящее исключение в неправильном файле.