Race condition in kafka benchmark driver

[travisdowns]

In the Kafka driver, we find the following code in initialize:

        if (config.reset) {
            // List existing topics
            ListTopicsResult result = admin.listTopics();
            try {
                Set<String> topics = result.names().get();
                // Delete all existing topics
                DeleteTopicsResult deletes = admin.deleteTopics(topics);
                deletes.all().get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
                throw new IOException(e);
            }
        }

This code will be called on every worker during initialization, near simultaneously. However, there is a race in that we list the topics, then delete this list. However, all workers will be competing to delete the same topics, so if worker A and B both read the list before either starts the delete operation, at least one of the two workers will fail with "unknown topics" exception because the other worker deleted the topic first.

Perhaps only the "ID 0" worker can execute this cleanup (though it does no seem that the workers know their ID) or something like that. This same code also exists in the MSK driver.