Me he enfrentado a este problema. Si los recursos son menores para que se inicie una SinkTask o SourceTask, esto puede suceder.
La memoria asignada al trabajador puede ser menor en algún momento. Por defecto, a los trabajadores se les asignan 250 MB. Por favor aumente esto. A continuación se muestra un ejemplo para asignar 2 GB de memoria para el trabajador que se ejecuta en modo distribuido.
KAFKA_HEAP_OPTS="-Xmx2G" sh $KAFKA_SERVICE_HOME/conectar-distribuido $KAFKA_CONFIG_HOME/conectar-avro-distribuido.propiedades
