Finding potential thread safety issues and race conditions in my multithreading code

Question

I am working on a project in which I have two tables in a different database with different schemas. So that means I have two different connection parameters for those two tables to connect using JDBC-

Let's suppose below is the config.property file.

NUMBER_OF_THREADS: 10
TOTAL_RUNNING_TIME: 2
SLEEP_TIME: 200
RANGE_VALID_USER_ID: 1-1000
RANGE_NON_EXISTENT_USER_ID: 10000-50000
PERCENTAGE_VALID_USER_ID: 95
FLAG_VALIDATE_DATA: false
FLAG_STOP_PROGRAM: false
TABLES: table1 table2
#For Table1
table1.url: jdbc:mysql://localhost:3306/garden
table1.user: gardener
table1.password: shavel
table1.driver: jdbc-driver
table1.percentage: 80
table1.columns: COL1,COL2,COl3,Col4,COL5
#For Table2
table2.url: jdbc:mysql://otherhost:3306/forest
table2.user: forester
table2.password: axe
table2.driver: jdbc-driver
table2.percentage: 20
table1.columns: COL10,COL11,COl12,Col13,COL14

Now from the above property files it means-

1. Number of Threads is 10
2. TOTAL_RUNNING_TIME means each thread will run for 2 minutes
3. Each thread will sleep for 200 milliseconds before making any other request
4. Range of valid User Id from which each thread will pick the id's depending on the percentage. Like 95% of each thread will pick Valid Id's and remaining 5% it will pick Non Existent User Id.
5. A boolean flag to validate the data.
6. A boolean flag to stop the program in case of any exceptions.

Below method will read the above config.properties file and make a TableConnectionInfo object for each tables.

private static void readPropertyFile() throws IOException {
 prop.load(LnPRead.class.getClassLoader().getResourceAsStream("config.properties"));
 threads = Integer.parseInt(prop.getProperty("NUMBER_OF_THREADS").trim());
 durationOfRun = Long.parseLong(prop.getProperty("TOTAL_RUNNING_TIME").trim());
 sleepTime = Long.parseLong(prop.getProperty("SLEEP_TIME").trim());
 flagValidateData = Boolean.parseBoolean(prop.getProperty("FLAG_VALIDATE_DATA").trim());
 flagTerminate = Boolean.parseBoolean(prop.getProperty("FLAG_STOP_PROGRAM").trim());
 startValidRange = Integer.parseInt(prop.getProperty("RANGE_VALID_USER_ID").trim().split("-")[0]);
 endValidRange = Integer.parseInt(prop.getProperty("RANGE_VALID_USER_ID").trim().split("-")[1]);
 startNonValidRange = Integer.parseInt(prop.getProperty("RANGE_NON_EXISTENT_USER_ID").trim().split("-")[0]);
 endNonValidRange = Integer.parseInt(prop.getProperty("RANGE_NON_EXISTENT_USER_ID").trim().split("-")[1]);
 percentageValidId = Double.parseDouble(prop.getProperty("PERCENTAGE_VALID_USER_ID").trim());
 tableNames = Arrays.asList(prop.getProperty("TABLES").trim().split(","));
 for (String arg : tableNames) {
 TableConnectionInfo ci = new TableConnectionInfo();
 ArrayList<String> columns = new ArrayList<String>();
 String url = prop.getProperty(arg + ".url").trim();
 String user = prop.getProperty(arg + ".user").trim();
 String password = prop.getProperty(arg + ".password").trim();
 String driver = prop.getProperty(arg + ".driver").trim();
 String table = prop.getProperty(arg + ".table").trim();
 double percentage = Double.parseDouble(prop.getProperty(arg + ".percentage").trim());
 columns = new ArrayList<String>(Arrays.asList(prop.getProperty(arg + ".columns").split(",")));
 ci.setUrl(url);
 ci.setUser(user);
 ci.setPassword(password);
 ci.setDriver(driver);
 ci.setTableName(table);
 ci.setPercentage(percentage);
 ci.setColumns(columns);
 tableList.put(arg, ci);
 }
}

Below is the TableConnectionInfo class that will hold all the table connection info for a particular table.

public class TableConnectionInfo {
 public String url;
 public String user;
 public String password;
 public String driver;
 public double percentage;
 public String tableName;
 public ArrayList<String> columns;
 public ArrayList<String> getColumns() {
 return columns;
 }
 public void setColumns(ArrayList<String> columns) {
 this.columns = columns;
 }
 public String getUrl() {
 return url;
 }
 public void setUrl(String url) {
 this.url = url;
 }
 public String getUser() {
 return user;
 }
 public void setUser(String user) {
 this.user = user;
 }
 public String getPassword() {
 return password;
 }
 public void setPassword(String password) {
 this.password = password;
 }
 public String getTableName() {
 return tableName;
 }
 public void setTableName(String tableName) {
 this.tableName = tableName;
 }
 public String getDriver() {
 return driver;
 }
 public void setDriver(String driver) {
 this.driver = driver;
 }
 public double getPercentage() {
 return percentage;
 }
 public void setPercentage(double percentage) {
 this.percentage = percentage;
 }
}

Now I am creating ExecutorService for specified number of threads and passing this tableList object (that I created by reading the config.property file) to constructor of Task class -

// create thread pool with given size
ExecutorService service = Executors.newFixedThreadPool(10);
long startTime = System.currentTimeMillis();
long endTime = startTime + (durationOfRun * 60 * 1000);
for (int i = 0; i < threads; i++) {
 service.submit(new Task(endTime, tableList));
}

Below is my Task class that implements Runnable interface in which each thread will make two connections for each table in the starting before doing anything meaningful.

class Task implements Runnable {
 private static final Logger LOG = Logger.getLogger(Task.class.getName());
 private static Random random = new SecureRandom();
 private Connection[] dbConnection = null;
 private PreparedStatement preparedStatement = null;
 private ResultSet rs = null;
 private HashMap<String, Connection> tableStatement = new HashMap<String, Connection>();
 private final long endTime;
 private final LinkedHashMap<String, TableConnectionInfo> tableLists;
 public static ConcurrentHashMap<Long, AtomicLong> selectHistogram = new ConcurrentHashMap<Long, AtomicLong>();
 public static ConcurrentHashMap<Long, AtomicLong> connectionHistogram = new ConcurrentHashMap<Long, AtomicLong>();
 public static ConcurrentHashMap<String, AtomicInteger> exceptionMap = new ConcurrentHashMap<String, AtomicInteger>();
 /**
 * Constructor to pass the values
 * 
 * @param durationOfRun
 * @param tableList
 */
 public Task(long endTime, LinkedHashMap<String, TableConnectionInfo> tableList) {
 this.endTime = endTime;
 this.tableLists = tableList;
 }
 @Override
 public void run() {
 try {
 int j = 0;
 dbConnection = new Connection[tableLists.size()];
 //loop around the map values and make the connection list
 for (TableConnectionInfo ci : tableLists.values()) {
 dbConnection[j] = getDBConnection(ci.getUrl(), ci.getUser(), ci.getPassword(), ci.getDriver());
 tableStatement.put(ci.getTableName(), dbConnection[j]);
 j++;
 }
 while (System.currentTimeMillis() <= endTime) {
 double randomNumber = random.nextDouble() * 100.0;
 TableConnectionInfo table = selectRandomTable(randomNumber);
 final int id = generateRandomId(random);
 final String columnsList = getColumns(table.getColumns());
 final String selectSql = "SELECT ID, CREATION_DATE, LAST_MODIFIED_DATE, " + columnsList + " from "
 + table.getTableName() + " where id = ?";
 preparedStatement = tableStatement.get(table.getTableName()).prepareCall(selectSql);
 preparedStatement.setString(1, String.valueOf(id));
 long start = System.nanoTime();
 rs = preparedStatement.executeQuery();
 long end = System.nanoTime() - start;
 final AtomicLong before = selectHistogram.putIfAbsent(end / 1000000L, new AtomicLong(1L));
 if (before != null) {
 before.incrementAndGet();
 }
 List<String> colData = new ArrayList<String>(columnsList.split(",").length);
 boolean foundData = false;
 if (rs.next()) {
 if (id >= LnPRead.startValidRange && id <= LnPRead.endValidRange) {
 foundData = true;
 for (String column : columnsList.split(",")) {
 colData.add(rs.getString(column.trim()));
 }
 } else {
 handleException(ReadConstants.NON_VALID_DATA, LnPRead.flagTerminate);
 }
 }
 if (LnPRead.flagValidateData && foundData) {
 for (String str : colData) {
 if (!isJSONValid(str, id)) {
 LOG.error("Invalid JSON String " + str + "with id" + id);
 handleException(ReadConstants.INVALID_JSON, LnPRead.flagTerminate);
 }
 }
 }
 if (LnPRead.sleepTime > 0L) {
 Thread.sleep(LnPRead.sleepTime);
 }
 }
 } catch (SQLException e) {
 handleException(e.getCause() != null ? e.getCause().toString() : e.toString(), LnPRead.flagTerminate);
 } catch (Exception e) {
 handleException(e.getCause() != null ? e.getCause().toString() : e.toString(), LnPRead.flagTerminate);
 } finally {
 closeConnection(preparedStatement, dbConnection);
 }
 }
 /**
 * A simple method to decide whether a String is a valid JSON String or not.
 * 
 * @param str
 * @param id
 * @return
 */
 public boolean isJSONValid(final String str, final int id) {
 boolean valid = true;
 try {
 final JSONObject obj = new JSONObject(str);
 final JSONArray data = obj.getJSONArray("lv");
 final int n = data.length();
 for (int i = 0; i < n; ++i) {
 final JSONObject lv = data.getJSONObject(i);
 JSONObject v = lv.getJSONObject("v");
 if (v.getInt("userId") != id) {
 valid = false;
 break;
 }
 }
 } catch (JSONException ex) {
 valid = false;
 }
 return valid;
 }
 /**
 * A simple method to get the column names in the order in which it was
 * inserted
 * 
 * @param columns
 * @return
 */
 private String getColumns(final List<String> columns) {
 List<String> copy = new ArrayList<String>(columns);
 Collections.shuffle(copy);
 int rNumber = random.nextInt(columns.size()) + 1;
 List<String> subList = copy.subList(0, rNumber);
 Collections.sort(subList, new Comparator<String>() {
 @Override
 public int compare(String o1, String o2) {
 return columns.indexOf(o1) < columns.indexOf(o2) ? -1 : 1;
 }
 });
 return StringUtils.join(subList, ",");
 }
 /**
 * A simple method to choose random table basis on the percentage
 * 
 * @param randomNumber
 * @return
 */
 private TableConnectionInfo selectRandomTable(double randomNumber) {
 double limit = 0;
 for (TableConnectionInfo ci : tableLists.values()) {
 limit += ci.getPercentage();
 if (randomNumber < limit) {
 return ci;
 }
 }
 throw new IllegalStateException();
 }
 /**
 * A simple method to decide which id's I need to pick either valid id or
 * non valid id
 * 
 * @param r
 * @return
 */
 private int generateRandomId(Random r) {
 int randomID;
 if (r.nextFloat() < LnPRead.percentageValidId / 100) {
 randomID = r.nextInt(LnPRead.endValidRange - LnPRead.startValidRange + 1)
 + LnPRead.startValidRange;
 } else {
 randomID = r.nextInt(LnPRead.endNonValidRange - LnPRead.startNonValidRange + 1)
 + LnPRead.startNonValidRange;
 }
 return randomID;
 }
 /**
 * A simple method to close the connection
 * 
 * @param callableStatement
 * @param dbConnection
 */
 private void closeConnection(PreparedStatement preparedStatement, Connection[] dbConnection) {
 if (preparedStatement != null) {
 try {
 preparedStatement.close();
 preparedStatement = null;
 } catch (SQLException e) {
 LOG.error("Threw a SQLException in finally block of prepared statement " + getClass().getSimpleName(), e);
 }
 }
 for (Connection con : dbConnection) {
 if (con != null) {
 try {
 con.close();
 con = null;
 } catch (SQLException e) {
 LOG.error("Threw a SQLException in finally block of dbConnection " + getClass().getSimpleName(), e);
 }
 }
 }
 }
 /**
 * Attempts to establish a connection to the given database URL
 * 
 * @return the db connection
 */
 private Connection getDBConnection(String url, String username, String password, String driver) {
 Connection dbConnection = null;
 try {
 Class.forName(driver);
 long start = System.nanoTime();
 dbConnection = DriverManager.getConnection(url, username, password);
 long end = System.nanoTime() - start;
 final AtomicLong before = connectionHistogram.putIfAbsent(end / 1000000, new AtomicLong(1L));
 if (before != null) {
 before.incrementAndGet();
 }
 } catch (ClassNotFoundException e) {
 handleException(e.getCause() != null ? e.getCause().toString() : e.toString(), LnPRead.flagTerminate);
 } catch (SQLException e) {
 handleException(e.getCause() != null ? e.getCause().toString() : e.toString(), LnPRead.flagTerminate);
 }
 return dbConnection;
 }
 /**
 * A simple method that will add the count of exceptions and name of
 * exception to a map
 * 
 * @param cause
 * @param flagTerminate
 */
 private static void handleException(String cause, boolean flagTerminate) {
 LOG.error(cause);
 AtomicInteger count = exceptionMap.get(cause);
 if (count == null) {
 count = new AtomicInteger();
 AtomicInteger curCount = exceptionMap.putIfAbsent(cause, count);
 if (curCount != null) {
 count = curCount;
 }
 }
 count.incrementAndGet();
 if (flagTerminate) {
 System.exit(1);
 }
 }
}

In my first try block in the run method, I am making dbConnection array for storing two different database connections. And then I have a tableStatement as HashMap in which I am storing corresponding table name with it's dbConnection. For example Table1 will have Table1 connection in the tableStatement HashMap.

And then after that I am applying this logic in the while loop so that each thread should run for 2 minutes-

/* Generate random number and check to see whether that random number
 * falls between 1 and 80, if yes, then choose table1
 * and then use table1 connection and statement that I made above and do a SELECT query on that table.
 * If that random numbers falls between 81 and 100 then choose table2 
 * and then use table2 connection and statement and do a SELECT query on that table
 */

1. Generating Random number between 1 and 100.
2. If that random number is less than table1.getPercentage() then I am choosing table1 and then use table1 connection object to make a SELECT sql call to that database. Else choose table2 and then use table2 connection object to make a SELECT SQL call to that database.
3. After executing SELECT SQL, if result set contains any data, then check to see whether the id is between Valid Range. If it is between the valid range then store it in colData list of String but if it not in the valid range, log an exception.
4. After that if the flag to validate the data is true, then validate the JSON String.

Problem Statement:

I am trying to see whether is there any potential thread safety issues or race conditions possible here in my run method? And is there anything that I can improve in my code?

I know, it's a lot of code that somebody has to go through. I am in the starting phase of multithreading so that is the reason I am having problem initially.

Answer 1

1. It looks thread-safe for me, I've not found any concurrency-related issue except that you might want to to use a ThreadLocalRandom or use separate SecureRandom instances with a ThreadLocal for performance reasons. See also. Is SecureRandom thread safe?

2. boolean valid = true;
   

   Instead of this local variable you could return immediately if you know the result:

   public boolean isJSONValid(final String str, final int id) {
    try {
    final JSONObject obj = new JSONObject(str);
    final JSONArray data = obj.getJSONArray("lv");
    final int n = data.length();
    for (int i = 0; i < n; ++i) {
    final JSONObject lv = data.getJSONObject(i);
    JSONObject v = lv.getJSONObject("v");
    if (v.getInt("userId") != id) {
    return false;
    }
    }
    } catch (JSONException ex) {
    return false;
    }
    return true;
   }
   

3. The following fields could be local variables in run:

   private Connection[] dbConnection = null;
   private PreparedStatement preparedStatement = null;
   private ResultSet rs = null;
   private HashMap<String, Connection> tableStatement = new HashMap<String, Connection>();
   

   Effective Java, Second Edition, Item 45: Minimize the scope of local variables

4. The null assignment is unnecessary here (in both cases):

   if (preparedStatement != null) {
    try {
    preparedStatement.close();
    preparedStatement = null;
    } catch (SQLException e) {
    LOG.error("Threw a SQLException in finally block of prepared statement " + getClass().getSimpleName(),
    e);
    }
   }
   for (Connection con: dbConnection) {
    if (con != null) {
    try {
    con.close();
    con = null;
    } catch (SQLException e) {
    LOG.error("Threw a SQLException in finally block of dbConnection " + getClass().getSimpleName(), e);
    }
    }
   }
   

   It only nulls a local variable/parameter which will be destroyed anyway at the end of the block/method. I guess you wanted to null the field, for that you need a this. prefix but as above, the fields could be local variables in run, so it's unnecessary.

5. A guard clause could make the second loop flatten:

   for (Connection con: dbConnection) {
    if (con == null) {
    continue;
    }
    try {
    con.close();
    } catch (SQLException e) {
    LOG.error("Threw a SQLException in finally block of dbConnection " + getClass().getSimpleName(), e);
    }
   }
   

6. LOG.error("Threw a SQLException in finally block of prepared statement " 
    + getClass().getSimpleName(), e);
   

   If you are using Logback or Log4j the getClass().getSimpleName() is unnecessary, you can set in the log4j.xml or logback.xml to log the class name for every log statement.

7. Instead of System.nanoTime(), you could use a stopwatch class, like Guava Stopwatch or Apache Commons StopWatch).

8. You could extract out the increment part of this method for higher abstraction level:

   private static void handleException(String cause, boolean flagTerminate) {
    LOG.error(cause);
    AtomicInteger count = exceptionMap.get(cause);
    if (count == null) {
    count = new AtomicInteger();
    AtomicInteger curCount = exceptionMap.putIfAbsent(cause, count);
    if (curCount != null) {
    count = curCount;
    }
    }
    count.incrementAndGet();
    if (flagTerminate) {
    System.exit(1);
    }
   }
   

   It's easier to read, it's easier to get an overview what the method does (without the details) and you can still check them if you are interested in. Furthermore, you need to read and understand less code (not the whole original method) if you want to modify just a small part of it.

   private static void handleException(String cause, boolean flagTerminate) {
    LOG.error(cause);
    incrementExceptionCount(cause);
    if (flagTerminate) {
    System.exit(1);
    }
   }
   private static void incrementExceptionCount(final String cause) {
    ...
   }
   

9. I guess you could use here the same structure with AtomicLongs:

   private static void incrementExceptionCount(final String cause) {
    AtomicInteger previousValue = exceptionMap.putIfAbsent(cause, new AtomicInteger());
    if (previousValue != null) {
    previousValue.incrementAndGet();
    }
   }
   

   You could also extract a similar method for AtomicLongs, since currently it's duplicated in the code.

10. Anyway, instead of the ConcurrentHashMap<Long, AtomicLong> structures I'd use Guava's AtomicLongMap. It was designed for these situations.

    (Effective Java, 2nd edition, Item 47: Know and use the libraries The author mentions only the JDK's built-in libraries but I think the reasoning could be true for other libraries too.)

11. The logic in the method parameter is duplicated here.

    } catch (ClassNotFoundException e) {
     handleException(e.getCause() != null ? e.getCause().toString() : e.toString(), LnPRead.flagTerminate);
    } catch (SQLException e) {
     handleException(e.getCause() != null ? e.getCause().toString() : e.toString(), LnPRead.flagTerminate);
    }
    

    It could be moved a new method:

    private void handleException2(Exception e) {
     handleException(e.getCause() != null ? e.getCause().toString() : e.toString(), LnPRead.flagTerminate);
    }
    

    A few explanatory local variable would make it more readable:

    private void handleException2(final Exception e) {
     final Throwable cause = e.getCause();
     final String exceptionString;
     if (cause == null) {
     exceptionString = e.toString();
     } else {
     exceptionString = cause.toString();
     }
     handleException(exceptionString, LnPRead.flagTerminate);
    }
    

    Reference: Chapter 6. Composing Methods, Introduce Explaining Variable in Refactoring: Improving the Design of Existing Code by Martin Fowler:

    Put the result of the expression, or parts of the expression, in a temporary variable with a name that explains the purpose.

    And Clean Code by Robert C. Martin, G19: Use Explanatory Variables.

12. The randomID local variable is unnecessary here:

    private int generateRandomId(Random r) {
     int randomID;
     if (r.nextFloat() < LnPRead.percentageValidId / 100) {
     randomID = r.nextInt(LnPRead.endValidRange - LnPRead.startValidRange + 1) + LnPRead.startValidRange;
     } else {
     randomID = r.nextInt(LnPRead.endNonValidRange - LnPRead.startNonValidRange + 1)
     + LnPRead.startNonValidRange;
     }
     return randomID;
    }
    

    You could return immediately (and use an explanatory variable too):

    private int generateRandomId(final Random r) {
     final boolean generateValidId = r.nextFloat() < LnPRead.percentageValidId / 100;
     if (generateValidId) {
     return r.nextInt(LnPRead.endValidRange - LnPRead.startValidRange + 1) + LnPRead.startValidRange;
     } else {
     return r.nextInt(LnPRead.endNonValidRange - LnPRead.startNonValidRange + 1)
     + LnPRead.startNonValidRange;
     }
    }
    

    Note that how easy to decide now whether the if or the else branch generates valid IDs. (I've just guessed here, you might need to invert the condition.)

13. I'd move the random.nextDouble() inside the selectRandomTable method.

    double randomNumber = random.nextDouble() * 100.0;
    TableConnectionInfo table = selectRandomTable(randomNumber);
    

    Nothing indicates the parameter should be between 0 and 100. It should be an internal part of the method to choose a proper random number.

14. In the same method, the IllegalStateException should contain some debugging information, like the generated random number. It would help debugging.

15. columnsList.split(",") could be extracted out to a local variable. It's used multiple times.

16. durationOfRun = Long.parseLong(prop.getProperty("TOTAL_RUNNING_TIME").trim());
    sleepTime = Long.parseLong(prop.getProperty("SLEEP_TIME").trim());
    

    The parsing and trimming could have a separate function:

    private static long loadLong(Properties properties, String key) {
     return Long.parseLong(properties.getProperty(key).trim());
    }
    

    You can create similar functions for the boolean and integer fields too.

17. for (String arg: tableNames) {
    

    arg is not too descriptive here. I'd call it tableName.

18. I'd rename isJSONValid to isJsonValid or isValidJson. I think it would be easier to read (although it's inconsistent with the JSON library's naming but I think that's a minor issue). From Effective Java, 2nd edition, Item 56: Adhere to generally accepted naming conventions:

    While uppercase may be more common, a strong argument can made in favor of capitalizing only the first letter: even if multiple acronyms occur back-to-back, you can still tell where one word starts and the next word ends. Which class name would you rather see, HTTPURL or HttpUrl?

19. The isJSONValid contains two magic numbers:

    JSONArray data = obj.getJSONArray("lv");
    int n = data.length();
    for (int i = 0; i < n; ++i) {
     JSONObject lv = data.getJSONObject(i);
     JSONObject v = lv.getJSONObject("v");
    ...
    

    They are the lv and v strings. Constants with longer and more descriptive name would help readers to figure out what should be the content of these fields. You could also rename the variable names too (data, lv, v) to express the intent.

20. final int n = data.length();
    for (int i = 0; i < n; ++i) {
    

    It seem a microoptimization, I guess the JVM will cache that for you if it counts. (JVM option to optimize loop statements)

21.  * @param str
     * @param id
     * @return
    

    I'd remove these comments. (Clean Code by Robert C. Martin: Chapter 4: Comments, Noise Comments)

22. public boolean isJSONValid(final String str, final int id) {
    

    I'd rename the parameter to jsonString from str.

23. private HashMap<String, Connection> tableStatement = new HashMap<String, Connection>();
    private final LinkedHashMap<String, TableConnectionInfo> tableLists; 
    

    HashMap<...> reference types should be simply Map<...>. See: Effective Java, 2nd edition, Item 52: Refer to objects by their interfaces

24. if (id >= LnPRead.startValidRange && id <= LnPRead.endValidRange) {
    

    startValidRange sounds like a method (the name starts with a verb/action). I'd call it validRangeStart.