kafka-server-stop.sh non funziona quando Kafka ha iniziato dallo script Python

Question

Dopo aver distribuito alcune istanze di Apache Kafka su nodi remoti, ho riscontrato un problema con lo script kafka-server-stop.sh che fa parte dell'archivio di Kafka.

Di default contiene:

#!/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 

e questo script funziona grande se eseguo apache Kafka non processo in background, ad esempio:

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

anche funziona quando eseguo come sfondo processo:

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

ma sui miei nodi remoti lo eseguo (con l'uso di Ans ible) con questo script python:

#!/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() 

dopo l'esecuzione

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

Zookeeper viene arrestato (come dovrebbe essere), ma Kafka è ancora in esecuzione.

Che cosa è più interessante, quando invoco (a mano)

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

o

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

kafka-server-stop.sh arresti correttamente esempio Kafka. Sospetto che questo problema possa essere causato da qualche cosa di Linux/Python.

Answer 1

La mia ipotesi: kafka-server-stop.sh utilizza tubi di shell. Quindi Popen avrebbe bisogno dell'argomento shell=True.

Vedi https://docs.python.org/2/library/subprocess.html#subprocess.Popen

Answer 2

Kafka ha bisogno per completare il processo di arresto prima che i guardiani dello zoo shutsdown.

Quindi avviare i rilevatori di zooke, quindi i broker riproveranno il processo di arresto.

Ho avuto un caso simile. Il problema era che la mia configurazione non stava aspettando l'arresto dei broker kafka.

Spero che questo aiuti qualcuno. Mi ci è voluto un po 'per capire ...