Kafka-server-stop.sh не работает, когда Kafka начал с Python script

После развертывания некоторых экземпляров Apache Kafka на удаленных узлах я обнаружил проблему с kafka-server-stop.sh script, которая является частью архива Кафки.

По умолчанию он содержит:

#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# 
#    http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}' | xargs kill -SIGTERM

и этот script отлично работает, если я выполняю apache kafka как не фоновый процесс, например:

/var/lib/kafka/bin/kafka-server-start.sh /var/lib/kafka/config/server.properties

также работает, когда я выполняю его как фоновый процесс:

/var/lib/kafka/bin/kafka-server-start.sh /var/lib/kafka/config/server.properties &

но на моих удаленных узлах я выполняю его (с использованием Ansible) с этим python script:

#!/usr/bin/env python
import argparse
import os
import subprocess

KAFKA_PATH = "/var/lib/kafka/"

def execute_command_pipe_output(command_to_call):
  return subprocess.Popen(command_to_call, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

def execute_command_no_output(command_to_call):
  with open(os.devnull, "w") as null_file:
    return subprocess.Popen(command_to_call, stdout=null_file, stderr=subprocess.STDOUT)  

def start_kafka(args):
  command_to_call = ["nohup"]
  command_to_call += [KAFKA_PATH + "bin/zookeeper-server-start.sh"]
  command_to_call += [KAFKA_PATH + "config/zookeeper.properties"]

  proc = execute_command_no_output(command_to_call)

  command_to_call = ["nohup"]
  command_to_call += [KAFKA_PATH + "bin/kafka-server-start.sh"]
  command_to_call += [KAFKA_PATH + "config/server.properties"]

  proc = execute_command_no_output(command_to_call)

def stop_kafka(args):
  command_to_call = [KAFKA_PATH + "bin/kafka-server-stop.sh"]

  proc = execute_command_pipe_output(command_to_call)
  for line in iter(proc.stdout.readline, b''):
    print line,

  command_to_call = [KAFKA_PATH + "bin/zookeeper-server-stop.sh"]

  proc = execute_command_pipe_output(command_to_call)
  for line in iter(proc.stdout.readline, b''):
    print line,


if __name__ == "__main__":
  parser = argparse.ArgumentParser(description="Starting Zookeeper and Kafka instances")
  parser.add_argument('action', choices=['start', 'stop'], help="action to take")

  args = parser.parse_args()

  if args.action == 'start':
    start_kafka(args)
  elif args.action == 'stop':
    stop_kafka(args)
  else:
    parser.print_help()

после выполнения

manage-kafka.py start
manage-kafka.py stop

Zookeeper отключается (как и должно быть), но Kafka все еще работает.

Что более интересно, когда я вызываю (вручную)

nohup /var/lib/kafka/bin/kafka-server-stop.sh

или

nohup /var/lib/kafka/bin/kafka-server-stop.sh &

kafka-server-stop.sh корректно завершает экземпляр Kafka. Я подозреваю, что эта проблема может быть вызвана некоторыми проблемами Linux/Python.

Ответы

Ответ 1

Кафке нужно закончить процесс выключения до того, как отключится zookeepers.

Итак, запустите zookeepers, а затем брокеров повторит процесс выключения.

У меня был похожий случай. Проблема заключалась в том, что моя конфигурация не ждала закрытия брокеров kafka.

Надеюсь, это поможет кому-то. Мне потребовалось некоторое время, чтобы понять...

Ответ 2

Я столкнулся с этой проблемой много, прежде чем выяснить грубый способ решить проблему. Итак, что произошло, Кафка внезапно закрылась, но порт все еще используется.

Выполните следующие действия:

  1. Найдите идентификатор процесса, запущенного на этом порту: lsof -t -i :YOUR_PORT_NUMBER. ## это для мака
  2. Убить этот процесс kill -9 process_id

Ответ 4

Пожалуйста, используйте kafka-server-stop.sh перед запуском инструмента управления kafka-zookeeper-stop.sh. Сначала он отключит сервер от zookeeper, а затем остановит самого zookeeper. Пожалуйста, подождите 3-4 секунды, прежде чем начать снова.