aws emr インスタンスで samza を使用していますが、常に次のような例外が発生します。誰か助けてもらえますか?:
org.apache.samza.SamzaException: メッセージの送信に失敗しました。例外: java.lang.IllegalStateException: プロデューサが閉じられた後は送信できません。org.apache.samza.system.kafka.KafkaSystemProducer$$anonfun$send$5.apply(KafkaSystemProducer.scala:120) で org.apache.samza.system.kafka.KafkaSystemProducer$$anonfun$send$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) org.apache.samza.task.TaskInstanceCollector.send(TaskInstanceCollector.scala:61) org.apache.samza.storage.kv.LoggedStore.putAll(LoggedStore.scala) :72) org.apache.samza.storage.kv.SerializedKeyValueStore で。