/*
 * Copyright 2013 by Cloudsoft Corp.
 *
 * Licensed 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.
 */
package brooklyn.entity.messaging.kafka;

import java.util.Map;

import brooklyn.config.ConfigKey;
import brooklyn.location.basic.SshMachineLocation;
import brooklyn.util.collections.MutableMap;

public class KafkaBrokerSshDriver extends AbstractfKafkaSshDriver implements KafkaBrokerDriver {

    public KafkaBrokerSshDriver(KafkaBrokerImpl entity, SshMachineLocation machine) {
        super(entity, machine);
    }

    @Override
    protected Map<String, Integer> getPortMap() {
        return MutableMap.of("kafkaPort", getKafkaPort());
    }

    @Override
    protected ConfigKey<String> getConfigTemplateKey() {
        return KafkaBroker.KAFKA_BROKER_CONFIG_TEMPLATE;
    }

    @Override
    protected String getConfigFileName() {
        return "server.properties";
    }

    @Override
    protected String getLaunchScriptName() {
        return "kafka-server-start.sh";
    }

    @Override
    protected String getProcessIdentifier() {
        return "kafka\\.Kafka";
    }

    @Override
    public Integer getKafkaPort() {
        return getEntity().getAttribute(KafkaBroker.KAFKA_PORT);
    }

}

@@ -785,16 +785,61 @@
 ableMap;
+%0Aimport brooklyn.util.jmx.jmxrmi.JmxRmiAgent;
 %0A%0Apublic
@@ -1747,10 +1747,532 @@
 %0A    %7D%0A%0A
+    @Override%0A    public Map%3CString, String%3E getShellEnvironment() %7B%0A        // kafka script sets JMX_PORT if it isn't set%0A        return MutableMap.%3CString, String%3E builder()%0A                .putAll(super.getShellEnvironment())%0A                // seems odd to pass RMI port here, as it gets assigned to com.sun.mgmt.jmx.port in kafka-run-class.sh%0A                // but RMI server port works, whereas JMX port does not%0A                .put(%22JMX_PORT%22, String.valueOf(getRmiServerPort()))%0A                .build();%0A    %7D%0A%0A
 %7D%0A

fix kafka so it sets jmx port correctly