/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * 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 org.wso2.carbon.sample.kafka;

import org.apache.log4j.Logger;

import java.io.File;

/**
 * KakfaUtil class contains utility functions for the Kafka receiver class
 */
public class KafkaUtil {

	private static Logger log = Logger.getLogger(KafkaUtil.class);

	static String sampleDirectoryPath = ".." + File.separator + ".." + File.separator + ".." + File.separator +
			"samples" + File.separator + "artifacts" + File.separator + "sampleNumber" + File.separator;

	/**
	 * This method will construct the directory path of the data file
	 *
	 * @param sampleNumber  Number of the sample which is running currently
	 * @param format        Format of the file (ex: csv, txt)
	 * @param topic         topic of the message to be sent (the data file should be named with the topic)
	 * @param filePath      file path if a sample if not running
	 *
	 */
	public static String getEventFilePath(String sampleNumber, String format, String topic, String filePath)
			throws Exception {
		if (sampleNumber != null && sampleNumber.length() == 0) {
			sampleNumber = null;
		}

		if (filePath != null && filePath.length() == 0) {
			filePath = null;
		}

		String resultingFilePath;
		if (filePath != null && sampleNumber == null) {
			resultingFilePath = filePath;
		} else if (filePath == null && sampleNumber != null) {
			if(format.equalsIgnoreCase("csv")){
				resultingFilePath = sampleDirectoryPath.replace("sampleNumber", sampleNumber)+ topic + ".csv";
			}else{
				resultingFilePath = sampleDirectoryPath.replace("sampleNumber", sampleNumber)+ topic + ".txt";
			}
		} else {
			throw new Exception("In sampleNumber:'" + sampleNumber + "' and filePath:'" + filePath
					+ "' one must be null and other not null");
		}
		File file = new File(resultingFilePath);
		if (!file.isFile()) {
			throw new Exception("'" + resultingFilePath + "' is not a file");
		}
		if (!file.exists()) {
			throw new Exception("file '" + resultingFilePath + "' does not exist");
		}
		return resultingFilePath;
	}
}

/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * 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 org.wso2.carbon.sample.kafka;

import org.apache.log4j.Logger;

import java.io.File;

/**
 * KakfaUtil class contains utility functions for the Kafka receiver class
 */
public class KafkaUtil {

	private static Logger log = Logger.getLogger(KafkaUtil.class);

	static String sampleDirectoryPath = ".." + File.separator + ".." + File.separator + ".." + File.separator +
			"samples" + File.separator + "artifacts" + File.separator + "sampleNumber" + File.separator;

	/**
	 * This method will construct the directory path of the data file
	 *
	 * @param sampleNumber  Number of the sample which is running currently
	 * @param format        Format of the file (ex: csv, txt)
	 * @param topic         topic of the message to be sent (the data file should be named with the topic)
	 * @param filePath      file path if a sample if not running
	 *
	 */
	public static String getEventFilePath(String sampleNumber, String format, String topic, String filePath)
			throws Exception {
		if (sampleNumber != null && sampleNumber.length() == 0) {
			sampleNumber = null;
		}

		if (filePath != null && filePath.length() == 0) {
			filePath = null;
		}

		String resultingFilePath;
		if (filePath != null && sampleNumber == null) {
			resultingFilePath = filePath;
		} else if (filePath == null && sampleNumber != null) {
			if("csv".equalsIgnoreCase(format)){
				resultingFilePath = sampleDirectoryPath.replace("sampleNumber", sampleNumber)+ topic + ".csv";
			}else{
				resultingFilePath = sampleDirectoryPath.replace("sampleNumber", sampleNumber)+ topic + ".txt";
			}
		} else {
			throw new Exception("In sampleNumber:'" + sampleNumber + "' and filePath:'" + filePath
					+ "' one must be null and other not null");
		}
		File file = new File(resultingFilePath);
		if (!file.isFile()) {
			throw new Exception("'" + resultingFilePath + "' is not a file");
		}
		if (!file.exists()) {
			throw new Exception("file '" + resultingFilePath + "' does not exist");
		}
		return resultingFilePath;
	}
}

Change `format.equalsIgnoreCase("csv")` --> `"csv".equalsIgnoreCase(format)`