title: Spring Batch DeployerPartitionerHandler Closing the Kafka producer with timeoutMillis = 30000 ms
body: <p>I'm using spring batch DeployerPartitionerHandler for remote partitioning and kafka to send the status of each worker node to a topic. However, in worker nodes kafka producer is closing at the runtime.</p>
  <p>KafkaConfig.java</p>
  <pre><code>    import java.util.HashMap;
      import java.util.Map;
      
      import org.apache.kafka.clients.consumer.ConsumerConfig;
      import org.apache.kafka.clients.producer.ProducerConfig;
      import org.apache.kafka.common.serialization.StringDeserializer;
      import org.apache.kafka.common.serialization.StringSerializer;
      import org.springframework.context.annotation.Bean;
      import org.springframework.context.annotation.Configuration;
      import org.springframework.kafka.annotation.EnableKafka;
      import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
      import org.springframework.kafka.core.ConsumerFactory;
      import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
      import org.springframework.kafka.core.DefaultKafkaProducerFactory;
      import org.springframework.kafka.core.KafkaTemplate;
      import org.springframework.kafka.core.ProducerFactory;
      import org.springframework.kafka.support.serializer.JsonDeserializer;
      import org.springframework.kafka.support.serializer.JsonSerializer;
      
      import com.project.Entity.WorkerStatus;
      
      @Configuration
      @EnableKafka
      public class KafkaConfig {
          
          @Bean
          public ProducerFactory&lt;String, WorkerStatus&gt; producerFactory(){
              Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
              
              config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;127.0.0.1:9092&quot;);
              config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
              config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
              
              return new DefaultKafkaProducerFactory&lt;&gt;(config);
          }
          
          @SuppressWarnings({ &quot;unchecked&quot;, &quot;rawtypes&quot; })
          @Bean
          public KafkaTemplate kafkaTemplate() {
              return new KafkaTemplate(producerFactory());
          }
          
          
          @Bean
          public ConsumerFactory&lt;String, WorkerStatus&gt; consumerFactory(){
              Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
              config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;127.0.0.1:9092&quot;);
              config.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;Status&quot;);
              config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
              config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
              
              return new DefaultKafkaConsumerFactory&lt;&gt;(config, new StringDeserializer(), new JsonDeserializer&lt;&gt;(WorkerStatus.class));
          }
          
          @Bean
          public ConcurrentKafkaListenerContainerFactory&lt;String, WorkerStatus&gt; kafkaListnerContainerFactory(){
              ConcurrentKafkaListenerContainerFactory&lt;String, WorkerStatus&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
              factory.setConsumerFactory(consumerFactory());
              return factory;
          }
          
      }
  </code></pre>
  <p>Producer.java</p>
  <pre><code>import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.beans.factory.annotation.Value;
  import org.springframework.context.annotation.Configuration;
  import org.springframework.context.annotation.Profile;
  import org.springframework.kafka.core.KafkaTemplate;
  import org.springframework.scheduling.annotation.EnableScheduling;
  import org.springframework.scheduling.annotation.Scheduled;
  
  import com.project.Entity.WorkerStatus;
  
  @Configuration
  @EnableScheduling
  @Profile(&quot;worker&quot;)
  public class StatusScheduler {
      
      @Autowired
      private KafkaTemplate&lt;String, WorkerStatus&gt; kafkaTemplate;
      
      @Value(&quot;${spring.cloud.task.name:default}&quot;)
      private String taskName;
      
      @Value(&quot;${spring.cloud.task.job-execution-id:5}&quot;)
      private String jobId;
      
      private static final String TOPIC = &quot;WorkerStatus&quot;;
      
      @Scheduled(fixedRate = 10000)
      void sendStatus() {
          String[] tasknameArray = taskName.split(&quot;:&quot;);
          WorkerStatus status = new WorkerStatus();
          status.setJobExecutionId(Integer.valueOf(jobId));
          status.setPartition(tasknameArray[tasknameArray.length-1]);
          kafkaTemplate.send(TOPIC, status);
          System.out.println(&quot;message sent&quot;);
      }
      
  </code></pre>
  <p>worker error log</p>
  <pre><code>2021-10-25 13:20:39.181  INFO 18260 --- [           main] com.project.SpringBatchApplication       : Started SpringBatchApplication in 66.441 seconds (JVM running for 70.718)
      2021-10-25 13:20:39.219 DEBUG 18260 --- [           main] o.s.c.t.r.support.SimpleTaskRepository   : Updating: TaskExecution with executionId=49 with the following {exitCode=0, endTime=Mon Oct 25 13:20:39 IST 2021, exitMessage='null', errorMessage='null'}
      2021-10-25 13:20:39.237  INFO 18260 --- [   scheduling-1] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: 
          acks = 1
          batch.size = 16384
          bootstrap.servers = [127.0.0.1:9092]
          buffer.memory = 33554432
          client.dns.lookup = use_all_dns_ips
          client.id = producer-1
          compression.type = none
          connections.max.idle.ms = 540000
          delivery.timeout.ms = 120000
          enable.idempotence = false
          interceptor.classes = []
          internal.auto.downgrade.txn.commit = true
          key.serializer = class org.apache.kafka.common.serialization.StringSerializer
          linger.ms = 0
          max.block.ms = 60000
          max.in.flight.requests.per.connection = 5
          max.request.size = 1048576
          metadata.max.age.ms = 300000
          metadata.max.idle.ms = 300000
          metric.reporters = []
          metrics.num.samples = 2
          metrics.recording.level = INFO
          metrics.sample.window.ms = 30000
          partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
          receive.buffer.bytes = 32768
          reconnect.backoff.max.ms = 1000
          reconnect.backoff.ms = 50
          request.timeout.ms = 30000
          retries = 2147483647
          retry.backoff.ms = 100
          sasl.client.callback.handler.class = null
          sasl.jaas.config = null
          sasl.kerberos.kinit.cmd = /usr/bin/kinit
          sasl.kerberos.min.time.before.relogin = 60000
          sasl.kerberos.service.name = null
          sasl.kerberos.ticket.renew.jitter = 0.05
          sasl.kerberos.ticket.renew.window.factor = 0.8
          sasl.login.callback.handler.class = null
          sasl.login.class = null
          sasl.login.refresh.buffer.seconds = 300
          sasl.login.refresh.min.period.seconds = 60
          sasl.login.refresh.window.factor = 0.8
          sasl.login.refresh.window.jitter = 0.05
          sasl.mechanism = GSSAPI
          security.protocol = PLAINTEXT
          security.providers = null
          send.buffer.bytes = 131072
          socket.connection.setup.timeout.max.ms = 127000
          socket.connection.setup.timeout.ms = 10000
          ssl.cipher.suites = null
          ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
          ssl.endpoint.identification.algorithm = https
          ssl.engine.factory.class = null
          ssl.key.password = null
          ssl.keymanager.algorithm = SunX509
          ssl.keystore.certificate.chain = null
          ssl.keystore.key = null
          ssl.keystore.location = null
          ssl.keystore.password = null
          ssl.keystore.type = JKS
          ssl.protocol = TLSv1.3
          ssl.provider = null
          ssl.secure.random.implementation = null
          ssl.trustmanager.algorithm = PKIX
          ssl.truststore.certificates = null
          ssl.truststore.location = null
          ssl.truststore.password = null
          ssl.truststore.type = JKS
          transaction.timeout.ms = 60000
          transactional.id = null
          value.serializer = class org.springframework.kafka.support.serializer.JsonSerializer
      
      2021-10-25 13:20:39.291  INFO 18260 --- [           main] o.s.m.s.b.SimpleBrokerMessageHandler     : Stopping...
      2021-10-25 13:20:39.291  INFO 18260 --- [           main] o.s.m.s.b.SimpleBrokerMessageHandler     : BrokerAvailabilityEvent[available=false, SimpleBrokerMessageHandler [org.springframework.messaging.simp.broker.DefaultSubscriptionRegistry@6f347d7]]
      2021-10-25 13:20:39.292  INFO 18260 --- [           main] o.s.m.s.b.SimpleBrokerMessageHandler     : Stopped.
      2021-10-25 13:20:39.292  INFO 18260 --- [           main] org.quartz.core.QuartzScheduler          : Scheduler quartzScheduler_$_NON_CLUSTERED paused.
      2021-10-25 13:20:39.328  INFO 18260 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Removing {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
      2021-10-25 13:20:39.328  INFO 18260 --- [           main] o.s.i.channel.PublishSubscribeChannel    : Channel 'BatchApp.errorChannel' has 0 subscriber(s).
      2021-10-25 13:20:39.329  INFO 18260 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : stopped bean '_org.springframework.integration.errorLogger'
      2021-10-25 13:20:39.409  INFO 18260 --- [   scheduling-1] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 2.7.1
      2021-10-25 13:20:39.410  INFO 18260 --- [   scheduling-1] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 61dbce85d0d41457
      2021-10-25 13:20:39.410  INFO 18260 --- [   scheduling-1] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1635148239407
      2021-10-25 13:20:39.418  INFO 18260 --- [           main] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-1] Closing the Kafka producer with timeoutMillis = 30000 ms.
      2021-10-25 13:20:39.431 ERROR 18260 --- [   scheduling-1] o.s.s.s.TaskUtils$LoggingErrorHandler    : Unexpected error occurred in scheduled task
      
      java.lang.IllegalStateException: Cannot perform operation after producer has been closed
          at org.apache.kafka.clients.producer.KafkaProducer.throwIfProducerClosed(KafkaProducer.java:893) ~[kafka-clients-2.7.1.jar!/:na]
          at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:902) ~[kafka-clients-2.7.1.jar!/:na]
          at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:886) ~[kafka-clients-2.7.1.jar!/:na]
          at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:864) ~[spring-kafka-2.7.8.jar!/:2.7.8]
          at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:580) ~[spring-kafka-2.7.8.jar!/:2.7.8]
          at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:380) ~[spring-kafka-2.7.8.jar!/:2.7.8]
          at com.project.Status.StatusScheduler.sendStatus(StatusScheduler.java:35) ~[classes!/:0.0.1-SNAPSHOT]
          at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
          at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:78) ~[na:na]
          at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
          at java.base/java.lang.reflect.Method.invoke(Method.java:567) ~[na:na]
          at org.springframework.scheduling.support.ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:84) ~[spring-context-5.3.12.jar!/:5.3.12]
          at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) ~[spring-context-5.3.12.jar!/:5.3.12]
          at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
          at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) ~[na:na]
          at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[na:na]
          at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130) ~[na:na]
          at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630) ~[na:na]
          at java.base/java.lang.Thread.run(Thread.java:831) ~[na:na]
      
      2021-10-25 13:20:39.447  INFO 18260 --- [           main] org.apache.kafka.common.metrics.Metrics  : Metrics scheduler closed
      2021-10-25 13:20:39.447  INFO 18260 --- [           main] org.apache.kafka.common.metrics.Metrics  : Closing reporter org.apache.kafka.common.metrics.JmxReporter
      2021-10-25 13:20:39.448  INFO 18260 --- [           main] org.apache.kafka.common.metrics.Metrics  : Metrics reporters closed
      2021-10-25 13:20:39.448  INFO 18260 --- [           main] o.a.kafka.common.utils.AppInfoParser     : App info kafka.producer for producer-1 unregistered
      2021-10-25 13:20:39.451  INFO 18260 --- [           main] o.s.s.quartz.SchedulerFactoryBean        : Shutting down Quartz Scheduler
      2021-10-25 13:20:39.451  INFO 18260 --- [           main] org.quartz.core.QuartzScheduler          : Scheduler quartzScheduler_$_NON_CLUSTERED shutting down.
      2021-10-25 13:20:39.451  INFO 18260 --- [           main] org.quartz.core.QuartzScheduler          : Scheduler quartzScheduler_$_NON_CLUSTERED paused.
      2021-10-25 13:20:39.451  INFO 18260 --- [           main] org.quartz.core.QuartzScheduler          : Scheduler quartzScheduler_$_NON_CLUSTERED shutdown complete.
      2021-10-25 13:20:39.452  INFO 18260 --- [           main] j.LocalContainerEntityManagerFactoryBean : Closing JPA EntityManagerFactory for persistence unit 'default'
      2021-10-25 13:20:39.466  INFO 18260 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Shutdown initiated...
      2021-10-25 13:20:39.483  INFO 18260 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Shutdown completed.
  </code></pre>
  <p>is there any problem with kafka configuration? because for master Kakfa producer is working file but for batch job worker nodes it is giving me error.</p>