我在 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.system.kafka.KafkaSystemProducer.send(KafkaSystemProducer.scala:86) 在 org.apache.samza.util.ExponentialSleepStrategy.run(ExponentialSleepStrategy.scala:81) 在 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) 在 org.apache.samza.storage.kv.SerializedKeyValueStore。