diff --git a/src/ray/gcs/redis_module/ray_redis_module.cc b/src/ray/gcs/redis_module/ray_redis_module.cc index 56ad71582..f832f9e18 100644 --- a/src/ray/gcs/redis_module/ray_redis_module.cc +++ b/src/ray/gcs/redis_module/ray_redis_module.cc @@ -311,7 +311,23 @@ int TableAppend_DoWrite(RedisModuleCtx *ctx, RedisModuleString **argv, int argc, // must be unique, or else we will have gaps in the log. // TODO(rkn): We need to get rid of this uniqueness requirement. We can // easily have multiple log events with the same message. - RAY_CHECK(flags == REDISMODULE_ZADD_ADDED) << "Appended a duplicate entry"; + if (flags != REDISMODULE_ZADD_ADDED) { + // The following code is a workaround to store the data at a new unique + // key. This is so redis doesn't crash (we currently have duplicate keys + // for error conditions, which get delivered via pubsub). + size_t len; + const char *id_str = RedisModule_StringPtrLen(id, &len); + RAY_LOG(INFO) << "Duplicate key: " << std::string(id_str, len); + // Store the value into a unique new key, just to keep track of it and + // make sure the log size grows. + std::string postfix = std::to_string(index); + RedisModuleString *new_id = + RedisString_Format(ctx, "%S:%b", id, postfix.data(), postfix.size()); + RedisModuleKey *new_key = OpenPrefixedKey( + ctx, prefix_str, new_id, REDISMODULE_READ | REDISMODULE_WRITE, mutated_key_str); + RedisModule_ZsetAdd(new_key, index, data, &flags); + RAY_CHECK(flags == REDISMODULE_ZADD_ADDED); + } return REDISMODULE_OK; } else { // The requested index did not match the current length of the log. Return diff --git a/test/runtest.py b/test/runtest.py index 3e2faadf4..44aab828a 100644 --- a/test/runtest.py +++ b/test/runtest.py @@ -2402,6 +2402,29 @@ def test_ray_setproctitle(shutdown_only): ray.get(unique_1.remote()) +def test_duplicate_error_messages(shutdown_only): + ray.init(num_cpus=0) + + driver_id = ray.ray_constants.NIL_JOB_ID.id() + error_data = ray.gcs_utils.construct_error_message(driver_id, "test", + "message", 0) + + # Push the same message to the GCS twice (they are the same because we + # do not include a timestamp). + + r = ray.worker.global_worker.redis_client + + r.execute_command("RAY.TABLE_APPEND", ray.gcs_utils.TablePrefix.ERROR_INFO, + ray.gcs_utils.TablePubsub.ERROR_INFO, driver_id, + error_data) + + # Before https://github.com/ray-project/ray/pull/3316 this would + # give an error + r.execute_command("RAY.TABLE_APPEND", ray.gcs_utils.TablePrefix.ERROR_INFO, + ray.gcs_utils.TablePubsub.ERROR_INFO, driver_id, + error_data) + + @pytest.mark.skipif( os.getenv("TRAVIS") is None, reason="This test should only be run on Travis.")