-----BEGIN PGP SIGNED MESSAGE----- Hash: SHA512 It's hard to say. However, calling `KafkaStreams#close()` in the uncaught exception handler is considered bad practice (using a timeout avoids the deadlock, but it would be better to just set a boolean flag and call `close()` from outside the handler.
I assume your input topic has enough partitions and there are enough tasks to utilize all 10 instances? You also don't need to set the `group.id` -- this config will be ignored anyway and the `application.id` is used as consumer `group.id`. I would try to inspect the log (if required at DEBUG level) to get more intel what the application is doing. - -Matthias On 2/23/20 1:09 PM, Michelle Francois wrote: > I tried to create a kafka streams application, but there was a bug > in my code. I believe there was a deadlock for my application and > whenever I tried to run an application instance with the same > StreamsConfig.APPLICATION_ID_CONFIG id it could not start. I had to > create another instance with different > StreamsConfig.APPLICATION_ID_CONFIG id in order for it to start. I > fixed the bug by adding: > >> kafkaStreams.setUncaughtExceptionHandler((Thread thread, >> Throwable throwable) -> { kafkaStreams.close(1000L, >> TimeUnit.MILLISECONDS); }); >> > and I had never the same problem of deadlock. > > *Nevertheless, these days my code had another bug, and the > execution got stuck in a while loop. * I used several times: > >> pkill -u myuser >> > > with no problems. However, now when I try to run 10 instances of my > streams application only a portion of them are initialized. I mean > that the method > >> public void init(ProcessorContext processorContext) >> > in my 10 processors is executed only to a portion of my 10 kafka > streams application instances. To be more specific, I run a > scriptfile > >> mvn exec:java -Dexec.mainClass="SiteApplication" >> -Dexec.args="Vectors1 VectorSink1 VectorSink2 >> VectorSourceNodeName1 VectorProcessorName1 StateStoreName1 >> si1-client si1-group si1-appid" & mvn exec:java >> -Dexec.mainClass="SiteApplication" -Dexec.args="Vectors2 >> VectorSink3 VectorSink4 VectorSourceNodeName2 >> VectorProcessorName2 StateStoreName2 si2-client si2-group >> si2-appid" & mvn exec:java -Dexec.mainClass="SiteApplication" >> -Dexec.args="Vectors3 VectorSink5 VectorSink6 >> VectorSourceNodeName3 VectorProcessorName3 StateStoreName3 >> si3-client si3-group si3-appid" & mvn exec:java >> -Dexec.mainClass="SiteApplication" -Dexec.args="Vectors4 >> VectorSink7 VectorSink8 VectorSourceNodeName4 >> VectorProcessorName4 StateStoreName4 si4-client si4-group >> si4-appid" & mvn exec:java -Dexec.mainClass="SiteApplication" >> -Dexec.args="Vectors5 VectorSink9 VectorSink10 >> VectorSourceNodeName5 VectorProcessorName5 StateStoreName5 >> si5-client si5-group si5-appid" & mvn exec:java >> -Dexec.mainClass="SiteApplication" -Dexec.args="Vectors6 >> VectorSink11 VectorSink12 VectorSourceNodeName6 >> VectorProcessorName6 StateStoreName6 si6-client si6-group >> si6-appid" & mvn exec:java -Dexec.mainClass="SiteApplication" >> -Dexec.args="Vectors7 VectorSink13 VectorSink14 >> VectorSourceNodeName7 VectorProcessorName7 StateStoreName7 >> si7-client si7-group si7-appid" & mvn exec:java >> -Dexec.mainClass="SiteApplication" -Dexec.args="Vectors8 >> VectorSink15 VectorSink16 VectorSourceNodeName8 >> VectorProcessorName8 StateStoreName8 si8-client si8-group >> si8-appid" & mvn exec:java -Dexec.mainClass="SiteApplication" >> -Dexec.args="Vectors9 VectorSink17 VectorSink18 >> VectorSourceNodeName9 VectorProcessorName9 StateStoreName9 >> si9-client si9-group si9-appid" & mvn exec:java >> -Dexec.mainClass="SiteApplication" -Dexec.args="Vectors10 >> VectorSink19 VectorSink20 VectorSourceNodeName10 >> VectorProcessorName10 StateStoreName10 si10-client si10-group >> si10-appid" & >> > > and usually only 7 processors are initialized out of ten (Each > application has the same topology consisting of one Processor). > > *Whenever I don't change the StreamsConfig.APPLICATION_ID_CONFIG id > the same instances are not initialized.* *Whenever I change the > StreamsConfig.APPLICATION_ID_CONFIG id of them all, the number of > instances that get initialized change (they are now 8 or 9 now > initalized according to the change of the > StreamsConfig.APPLICATION_ID_CONFIG and they stick to that until I > change the StreamsConfig.APPLICATION_ID_CONFIG again to all of > them).* > > > My streams topology is*:* > > > public class SiteApplication { >> static Logger logger = >> Logger.getLogger(SiteApplication.class.getName()); >> >> >> >> >> public static void main(String[] args) throws Exception { >> >> PropertyConfigurator.configure(SiteApplication.class.getClassLoader() .getResourceAsStream("log4j.properties")); >> >> >> Serde<String> stringSerde = Serdes.String(); >> Serde<Integer> integerSerde = Serdes.Integer(); Serde<byte[]> >> byteArraySerde = Serdes.ByteArray(); >> >> Serializer<String> stringSerializer = stringSerde.serializer(); >> Deserializer<String> stringDeserializer = >> stringSerde.deserializer(); >> >> Serializer<Integer> integerSerializer = >> integerSerde.serializer(); Deserializer<Integer> >> integerDeserializer = integerSerde.deserializer(); >> >> Serializer<byte[]> byteArraySerializer = >> byteArraySerde.serializer(); Deserializer<byte[]> >> byteArrayDeserializer = byteArraySerde.deserializer(); >> >> Serializer<byte[]> byteArraySerializer1 = >> byteArraySerde.serializer(); Deserializer<byte[]> >> byteArrayDeserializer1 = byteArraySerde.deserializer(); >> >> >> Topology toplogy = new Topology(); >> >> String vectorSink1 = args[1]; String vectorSink2 = args[2]; >> String vectorSourceNodeName = args[3]; String vectorProcessorName >> = args[4]; String firstStateStoreName = args[5]; >> >> >> MyCustomStoreBuilder vectorStoreBuilder = new >> MyCustomStoreBuilder(firstStateStoreName); >> >> >> toplogy.addSource(vectorSourceNodeName, stringDeserializer, >> integerDeserializer, args[0]) .addProcessor(vectorProcessorName, >> ()->new SiteProcessor(firstStateStoreName, vectorSink1, >> vectorSink2), vectorSourceNodeName) >> .addSink(vectorSink1,"IncreaseOfC", stringSerializer, >> byteArraySerializer, vectorProcessorName) .addSink(vectorSink2, >> "Messages", stringSerializer, byteArraySerializer1, >> vectorProcessorName) .addStateStore(vectorStoreBuilder, >> vectorProcessorName); >> >> >> Properties props = new Properties(); >> props.put(StreamsConfig.CLIENT_ID_CONFIG, args[6]); >> props.put(ConsumerConfig.GROUP_ID_CONFIG, args[7]); >> props.put(StreamsConfig.APPLICATION_ID_CONFIG, args[8]); >> props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, " >> clu1.myuniversity.com:6667,clu2.myuniversity:6667, >> clu3.myuniversity.com:6667,clu4.myuniversity.com:6667"); >> props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 1); >> props.put(StreamsConfig.STATE_DIR_CONFIG, "/home/myuser/state"); >> >> props.put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RES ET_CONFIG), >> >> "earliest"); >> props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, >> Serdes.String().getClass() ); >> props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, >> Serdes.String().getClass() ); >> >> >> >> props.put(StreamsConfig.consumerPrefix(ConsumerConfig.METADATA_MAX_AG E_CONFIG), >> >> 60000); >> props.put(StreamsConfig.producerPrefix(ProducerConfig.METADATA_MAX_AG E_CONFIG), >> >> 60000); >> >> KafkaStreams kafkaStreams = new KafkaStreams(toplogy, props); >> kafkaStreams.setUncaughtExceptionHandler((Thread thread, >> Throwable throwable) -> { // here you should examine the >> throwable/exception and perform an appropriate action! >> logger.info("There is an error in site " + args[6]); >> kafkaStreams.close(1000L, TimeUnit.MILLISECONDS); }); >> >> >> logger.info("Starting FGM Application now for " + args[6]); >> kafkaStreams.cleanUp(); kafkaStreams.start(); >> >> // Add shutdown hook to respond to SIGTERM and gracefully close >> Kafka Streams Runtime.getRuntime().addShutdownHook(new >> Thread(()->{ try { System.out.println("Shutting down FGM >> Application now"); kafkaStreams.close(1000L, >> TimeUnit.MILLISECONDS); } catch (final Exception e) { >> e.printStackTrace(); } })); } >> >> >> >> } > > > > > > I have to say something to the administrator of the cluster of my > university running the kafka Streams but I do not know what > exactly is happening. > > *I do not know if It is a bug in my code because it is not stuck in > a while loop any more, because I run a steady version of my code > which ran in the past without problems.* > > *The problems began after being stuck in a while loop in each > application instance but I fixed this.* > > *Does the installation of the Apache Kafka need a restart by the > administrator or another change to configuration?* > > Is it a deadlock that I am missing? > > If it were a deadlock why it would not happen to all the > application instances but instead only to a portion of them? > > Thanks in advance. > -----BEGIN PGP SIGNATURE----- iQIzBAEBCgAdFiEEI8mthP+5zxXZZdDSO4miYXKq/OgFAl5fLMQACgkQO4miYXKq /Og8KhAArJyvbd8TcCTY1lBbgd+NigqJSOA153fXaP0UDDiAFl8bDBQQ3MziL6uz etpFABw5C/B5izX4OTn233M6hZXZaT2+xNd7i3z3aLkXUvmQNZJvsZxWz+feZAuT KQDQoSKzgdwuwyjKF2BHfrJCYd+fLjgAFTGTXBjIiJHhOUnM2rkSVuzX0PkZLRR0 6f0Otz5s8+BdyLyTfWdXcsF0y5rUl9a4dlOPHqt+LMCtDfJNQkAM204TzDBlisFq orJhcTLWPNC/UUKhe5RHGh8BY2XQBThwEIGw3T6VNlERQifZZgHOUJFfJUgN0Jps 1ejaTWWA/UETy89Jp8d93hJT3mSh46JIq/ylykiYVtfga2F80WfKBpwVEE/oYdmQ CbPWz2ZpMrJAHwYmo/dAayiMIv2Ux36A7NtmVr87r1oeb+euRzC/58p1yGBrNoGv xyU/bYbUbvKwoMxwE1yox9fK/bKqTfvJ4x7Ov3um49oG53fkUGcjzEiIfAEPiP5m +autPY6Q7oGErPQYPfraVVqfh14N3dx/KWP2gAsFcs1kvv8xlsNREJWcu/rHyLP8 K8oP+Jwe50qZesyvOco+GL/okkVGkp0DjJnQgKhKrTDuS14A6ZtY9t6/iSyKkgge RW332D6Vj+V6B/yzELACRAjJvt6dctQSirvJ+1KsRkdsiVx9VjQ= =ZzZl -----END PGP SIGNATURE-----