KafkaStreamsChannel : be-engine throws MissingSourceTopicException on startup.

Version Information

What is the expected behavior?

The kafka-streams engine from KafkaStreamsChannel project should start successfully.

What is the actual behavior?

The kafka-streams engine throws exception on startup as described below :
stream-thread [be_kafka_count_streams-d1299413-3206-4f76-92a0-71b7bd3fd846-StreamThread-1-consumer] Caught an error in the task assignment. Returning an error assignment.
org.apache.kafka.streams.errors.MissingSourceTopicException: Missing source topics.
at org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.prepareRepartitionTopics(StreamsPartitionAssignor.java:522) ~[kafka-streams-1.0.0.jar:?]
at org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.assign(StreamsPartitionAssignor.java:388) [kafka-streams-1.0.0.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeaderElected(ConsumerCoordinator.java:698) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.onLeaderElected(AbstractCoordinator.java:736) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.access$1000(AbstractCoordinator.java:112) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:640) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:603) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:1270) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:1245) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:206) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:169) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:129) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:617) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:427) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:312) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:251) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1323) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1254) [kafka-clients.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1227) [kafka-clients.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:1007) [kafka-streams-1.0.0.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.pollPhase(StreamThread.java:955) [kafka-streams-1.0.0.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:762) [kafka-streams-1.0.0.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:613) [kafka-streams-1.0.0.jar:?]
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:575) [kafka-streams-1.0.0.jar:?]

Please provide a unit test that demonstrates the bug.

1. Create the Kafka Streams channel jar from repo. The jar is automatically added in BE_HOME/lib/ext/tpcl.
2. Restart BE Studio and create channel of type Kafka Streams in BE project. Set all the fields in channel.
3. Configure the project acc, and build the EAR file.
4. Start kafka-streams engine using CDD and EAR, which throws exception as described above.

Please provide log files.

BE630v161.zip
KafkaStreamsChannel.zip
BE630-Kafka3.5.0.log