Kafka input crashing logstash on attempted group join during rebalance

[justinainsworth]

• Version:
  logstash-integration-kafka-10.7.0-java
• Operating System:
  Linux 5.4.0-1032-azure

Receiving the following error randomly on some of my logstash hosts. When it occurs the instance of logstash crashes and exits.

Dec 11 23:14:38 XX logstash[3034]: [2020-12-11T23:14:38,087][FATAL][logstash.runner ] An unexpected error occurred! {:error=>org.apache.kafka.common.KafkaException: Unexpected error in join group response: The group is rebalancing, so a rejoin is needed., :backtrace=>["org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:653)", "org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:574)", "org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:1096)", "org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:1076)", "org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(org/apache/kafka/clients/consumer/internals/RequestFuture.java:204)", "org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(org/apache/kafka/clients/consumer/internals/RequestFuture.java:167)", "org.apache.kafka.clients.consumer.internals.RequestFuture.complete(org/apache/kafka/clients/consumer/internals/RequestFuture.java:127)", "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java:599)", "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java:409)", "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java:294)", "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java:233)", "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java:212)", "org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:418)", "org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:358)", "org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:497)", "org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(org/apache/kafka/clients/consumer/KafkaConsumer.java:1274)", "org.apache.kafka.clients.consumer.KafkaConsumer.poll(org/apache/kafka/clients/consumer/KafkaConsumer.java:1238)", "org.apache.kafka.clients.consumer.KafkaConsumer.poll(org/apache/kafka/clients/consumer/KafkaConsumer.java:1173)", "jdk.internal.reflect.GeneratedMethodAccessor67.invoke(jdk/internal/reflect/GeneratedMethodAccessor67)", "jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(jdk/internal/reflect/DelegatingMethodAccessorImpl.java:43)", "java.lang.reflect.Method.invoke(java/lang/reflect/Method.java:566)", "org.jruby.javasupport.JavaMethod.invokeDirectWithExceptionHandling(org/jruby/javasupport/JavaMethod.java:456)", "org.jruby.javasupport.JavaMethod.invokeDirect(org/jruby/javasupport/JavaMethod.java:317)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_kafka_minus_10_dot_7_dot_0_minus_java.lib.logstash.inputs.kafka.thread_runner(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.7.0-java/lib/logstash/inputs/kafka.rb:283)", "org.jruby.RubyProc.call(org/jruby/RubyProc.java:318)", "java.lang.Thread.run(java/lang/Thread.java:834)"]}
Dec 11 23:14:38 XX logstash[3034]: [2020-12-11T23:14:38,788][ERROR][org.logstash.Logstash ] java.lang.IllegalStateException: Logstash stopped processing because of an error: (SystemExit) exit