diff --git a/firestore/CMakeLists.txt b/firestore/CMakeLists.txt
index 1d9c9c5f07..feaf22d323 100644
--- a/firestore/CMakeLists.txt
+++ b/firestore/CMakeLists.txt
@@ -138,6 +138,8 @@ set(android_SRCS
     src/android/wrapper.h
     src/android/write_batch_android.cc
     src/android/write_batch_android.h
+    src/jni/arena_ref.cc
+    src/jni/arena_ref.h
     src/jni/array.h
     src/jni/array_list.cc
     src/jni/array_list.h
diff --git a/firestore/integration_test_internal/CMakeLists.txt b/firestore/integration_test_internal/CMakeLists.txt
index 0205e5b164..1e9469e3fc 100644
--- a/firestore/integration_test_internal/CMakeLists.txt
+++ b/firestore/integration_test_internal/CMakeLists.txt
@@ -131,6 +131,7 @@ set(FIREBASE_INTEGRATION_TEST_ANDROID_TEST_SRCS
     src/android/snapshot_metadata_android_test.cc
     src/android/timestamp_android_test.cc
     src/android/transaction_options_android_test.cc
+    src/jni/arena_ref_test.cc
     src/jni/declaration_test.cc
     src/jni/env_test.cc
     src/jni/object_test.cc
diff --git a/firestore/integration_test_internal/src/document_reference_test.cc b/firestore/integration_test_internal/src/document_reference_test.cc
index cffda408ab..6ecb257a05 100644
--- a/firestore/integration_test_internal/src/document_reference_test.cc
+++ b/firestore/integration_test_internal/src/document_reference_test.cc
@@ -59,7 +59,7 @@ TEST_F(DocumentReferenceTest, RecoverFirestore) {
   DocumentReference doc = Document();
   ASSERT_EQ(db, doc.firestore());  // Sanity check
 
-  jni::Object doc_java = GetInternal(doc)->ToJava();
+  jni::Local<jni::Object> doc_java = GetInternal(doc)->ToJava();
   result = DocumentReferenceInternal::Create(env, doc_java);
   ASSERT_EQ(db, result.firestore());
 }
diff --git a/firestore/integration_test_internal/src/field_value_test.cc b/firestore/integration_test_internal/src/field_value_test.cc
index bd82c1aa90..7f75a15eb1 100644
--- a/firestore/integration_test_internal/src/field_value_test.cc
+++ b/firestore/integration_test_internal/src/field_value_test.cc
@@ -367,5 +367,12 @@ TEST_F(FieldValueTest, TestIncrementChoosesTheCorrectType) {
   // clang-format on
 }
 
+TEST_F(FieldValueTest, TestArenaRefMinimunLimit) {
+  std::vector<FieldValue> numbers;
+  for (size_t i = 0U; i < 60000; i++) {
+    numbers.push_back(FieldValue::Integer(i));
+  }
+}
+
 }  // namespace firestore
 }  // namespace firebase
diff --git a/firestore/integration_test_internal/src/jni/arena_ref_test.cc b/firestore/integration_test_internal/src/jni/arena_ref_test.cc
new file mode 100644
index 0000000000..d621e765a7
--- /dev/null
+++ b/firestore/integration_test_internal/src/jni/arena_ref_test.cc
@@ -0,0 +1,278 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * 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.
+ */
+
+#include "firestore/src/jni/arena_ref.h"
+#include "firestore/src/android/firestore_android.h"
+#include "firestore/src/common/make_unique.h"
+#include "firestore/src/jni/loader.h"
+#include "firestore/src/jni/long.h"
+
+#include "firestore_integration_test.h"
+
+#include "gtest/gtest.h"
+
+namespace firebase {
+namespace firestore {
+namespace jni {
+namespace {
+
+constexpr char kException[] = "java/lang/Exception";
+Method<String> kConstructor("<init>", "(Ljava/lang/String;)V");
+
+class ArenaRefTestAndroid : public FirestoreIntegrationTest {
+ public:
+  ArenaRefTestAndroid() : loader_(app()), env_(make_unique<Env>(GetEnv())) {
+    loader_.LoadClass(kException);
+    loader_.Load(kConstructor);
+  }
+
+  ~ArenaRefTestAndroid() override {
+    // Ensure that after the test is done that any pending exception is cleared
+    // to prevent spurious errors in the teardown of FirestoreIntegrationTest.
+    env_->ExceptionClear();
+  }
+
+  Env& env() const { return *env_; }
+
+  void throwException() {
+    Local<Class> clazz = env().FindClass(kException);
+    jmethodID ctor =
+        env().GetMethodId(clazz, "<init>", "(Ljava/lang/String;)V");
+
+    Local<String> message = env().NewStringUtf("Testing throw");
+    Local<Throwable> exception = env().New<Throwable>(clazz, ctor, message);
+    // After throwing, use EXPECT rather than ASSERT to ensure that the
+    // exception is cleared.
+    env().Throw(exception);
+    EXPECT_FALSE(env().ok());
+  }
+
+  void clearExceptionOccurred() {
+    Local<Throwable> thrown = env().ClearExceptionOccurred();
+    EXPECT_EQ(thrown.GetMessage(env()), "Testing throw");
+  }
+
+ private:
+  // Env is declared as having a `noexcept(false)` destructor, which causes the
+  // compiler to propagate this into `EnvTest`'s destructor, but this conflicts
+  // with the declaration of the parent class. Holding `Env` with a unique
+  // pointer sidesteps this restriction.
+  std::unique_ptr<Env> env_;
+  Loader loader_;
+};
+
+TEST_F(ArenaRefTestAndroid, DefaultConstructor) {
+  ArenaRef arena_ref;
+  EXPECT_EQ(arena_ref.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, ConstructsFromNull) {
+  Local<String> string;
+  ArenaRef arena_ref(env(), string);
+  EXPECT_EQ(arena_ref.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, ConstructsFromValid) {
+  Local<String> string = env().NewStringUtf("hello world");
+  ArenaRef arena_ref(env(), string);
+  EXPECT_TRUE(env().IsSameObject(arena_ref.get(env()), string));
+}
+
+TEST_F(ArenaRefTestAndroid, CopyConstructsFromNull) {
+  ArenaRef arena_ref1;
+  ArenaRef arena_ref2(arena_ref1);
+  EXPECT_EQ(arena_ref2.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, CopyConstructsFromValid) {
+  Local<String> string = env().NewStringUtf("hello world");
+
+  ArenaRef arena_ref1(env(), string);
+  ArenaRef arena_ref2(arena_ref1);
+  EXPECT_TRUE(env().IsSameObject(arena_ref1.get(env()), string));
+  EXPECT_TRUE(env().IsSameObject(arena_ref2.get(env()), string));
+}
+
+TEST_F(ArenaRefTestAndroid, CopyAssignsFromNullToNull) {
+  ArenaRef arena_ref1, arena_ref2;
+  arena_ref2 = arena_ref1;
+  EXPECT_EQ(arena_ref1.get(env()).get(), nullptr);
+  EXPECT_EQ(arena_ref2.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, CopyAssignsFromNullToValid) {
+  Local<String> string = env().NewStringUtf("hello world");
+
+  ArenaRef arena_ref1;
+  ArenaRef arena_ref2(env(), string);
+  arena_ref2 = arena_ref1;
+  EXPECT_EQ(arena_ref1.get(env()).get(), nullptr);
+  EXPECT_EQ(arena_ref2.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, CopyAssignsFromValidToNull) {
+  Local<String> string = env().NewStringUtf("hello world");
+
+  ArenaRef arena_ref1;
+  ArenaRef arena_ref2(env(), string);
+  arena_ref1 = arena_ref2;
+  EXPECT_TRUE(env().IsSameObject(arena_ref1.get(env()), string));
+  EXPECT_TRUE(env().IsSameObject(arena_ref2.get(env()), string));
+}
+
+TEST_F(ArenaRefTestAndroid, CopyAssignsFromValidToValid) {
+  Local<String> string1 = env().NewStringUtf("hello world");
+  Local<String> string2 = env().NewStringUtf("hello earth");
+
+  ArenaRef arena_ref1(env(), string1);
+  ArenaRef arena_ref2(env(), string2);
+  arena_ref1 = arena_ref2;
+
+  EXPECT_TRUE(env().IsSameObject(arena_ref1.get(env()), string2));
+  EXPECT_TRUE(env().IsSameObject(arena_ref2.get(env()), string2));
+}
+
+TEST_F(ArenaRefTestAndroid, CopyAssignsFromNullObjectItself) {
+  ArenaRef arena_ref1;
+  arena_ref1 = arena_ref1;
+  EXPECT_EQ(arena_ref1.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, CopyAssignsFromValidObjectItself) {
+  Local<String> string1 = env().NewStringUtf("hello world");
+
+  ArenaRef arena_ref1(env(), string1);
+  arena_ref1 = arena_ref1;
+  EXPECT_TRUE(env().IsSameObject(arena_ref1.get(env()), string1));
+}
+
+TEST_F(ArenaRefTestAndroid, MoveConstructsFromNull) {
+  ArenaRef arena_ref1;
+  ArenaRef arena_ref2(std::move(arena_ref1));
+  EXPECT_EQ(arena_ref2.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, MoveConstructsFromValid) {
+  Local<String> string = env().NewStringUtf("hello world");
+
+  ArenaRef arena_ref2(env(), string);
+  ArenaRef arena_ref3(std::move(arena_ref2));
+  EXPECT_TRUE(env().IsSameObject(arena_ref3.get(env()), string));
+}
+
+TEST_F(ArenaRefTestAndroid, MoveAssignsFromNullToNull) {
+  ArenaRef arena_ref1, arena_ref2;
+  arena_ref2 = std::move(arena_ref1);
+  EXPECT_EQ(arena_ref2.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, MoveAssignsFromNullToValid) {
+  Local<String> string = env().NewStringUtf("hello world");
+
+  ArenaRef arena_ref1;
+  ArenaRef arena_ref2(env(), string);
+  arena_ref2 = std::move(arena_ref1);
+  EXPECT_EQ(arena_ref2.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, MoveAssignsFromValidToNull) {
+  Local<String> string = env().NewStringUtf("hello world");
+
+  ArenaRef arena_ref1;
+  ArenaRef arena_ref2(env(), string);
+  arena_ref1 = std::move(arena_ref2);
+  EXPECT_TRUE(env().IsSameObject(arena_ref1.get(env()), string));
+}
+
+TEST_F(ArenaRefTestAndroid, MoveAssignsFromValidToValid) {
+  Local<String> string1 = env().NewStringUtf("hello world");
+  Local<String> string2 = env().NewStringUtf("hello earth");
+
+  ArenaRef arena_ref1(env(), string1);
+  ArenaRef arena_ref2(env(), string2);
+  arena_ref1 = std::move(arena_ref2);
+  EXPECT_TRUE(env().IsSameObject(arena_ref1.get(env()), string2));
+}
+
+TEST_F(ArenaRefTestAndroid, MoveAssignsFromNullObjectItself) {
+  ArenaRef arena_ref1;
+  arena_ref1 = std::move(arena_ref1);
+  EXPECT_EQ(arena_ref1.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, MoveAssignsFromValidObjectItself) {
+  Local<String> string1 = env().NewStringUtf("hello world");
+
+  ArenaRef arena_ref1(env(), string1);
+  arena_ref1 = std::move(arena_ref1);
+  EXPECT_TRUE(env().IsSameObject(arena_ref1.get(env()), string1));
+}
+
+TEST_F(ArenaRefTestAndroid, ThrowBeforeConstruct) {
+  Local<String> string = env().NewStringUtf("hello world");
+  EXPECT_EQ(string.ToString(env()).size(), 11U);
+  throwException();
+  ArenaRef arena_ref(env(), string);
+  clearExceptionOccurred();
+  EXPECT_EQ(arena_ref.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, ThrowBeforeCopyConstruct) {
+  Local<String> string = env().NewStringUtf("hello world");
+  ArenaRef arena_ref1(env(), string);
+  EXPECT_EQ(arena_ref1.get(env()).ToString(env()).size(), 11U);
+  throwException();
+  ArenaRef arena_ref2(arena_ref1);
+  clearExceptionOccurred();
+  EXPECT_EQ(arena_ref2.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, ThrowBeforeCopyAssignment) {
+  Local<String> string = env().NewStringUtf("hello world");
+  ArenaRef arena_ref1(env(), string);
+  ArenaRef arena_ref2;
+  EXPECT_EQ(arena_ref1.get(env()).ToString(env()).size(), 11U);
+  throwException();
+  arena_ref2 = arena_ref1;
+  clearExceptionOccurred();
+  EXPECT_EQ(arena_ref2.get(env()).get(), nullptr);
+}
+
+TEST_F(ArenaRefTestAndroid, ThrowBeforeMoveConstruct) {
+  Local<String> string = env().NewStringUtf("hello world");
+  ArenaRef arena_ref1(env(), string);
+  EXPECT_EQ(arena_ref1.get(env()).ToString(env()).size(), 11U);
+  throwException();
+  ArenaRef arena_ref2(std::move(arena_ref1));
+  clearExceptionOccurred();
+  EXPECT_TRUE(env().IsSameObject(arena_ref2.get(env()), string));
+}
+
+TEST_F(ArenaRefTestAndroid, ThrowBeforeMoveAssignment) {
+  Local<String> string = env().NewStringUtf("hello world");
+  ArenaRef arena_ref1(env(), string);
+  ArenaRef arena_ref2;
+  EXPECT_EQ(arena_ref1.get(env()).ToString(env()).size(), 11U);
+  throwException();
+  arena_ref2 = std::move(arena_ref1);
+  clearExceptionOccurred();
+  EXPECT_TRUE(env().IsSameObject(arena_ref2.get(env()), string));
+}
+
+}  // namespace
+}  // namespace jni
+}  // namespace firestore
+}  // namespace firebase
diff --git a/firestore/integration_test_internal/src/jni/env_test.cc b/firestore/integration_test_internal/src/jni/env_test.cc
index d31992e48e..e33b09b552 100644
--- a/firestore/integration_test_internal/src/jni/env_test.cc
+++ b/firestore/integration_test_internal/src/jni/env_test.cc
@@ -128,6 +128,18 @@ TEST_F(EnvTest, CallsVoidMethods) {
   EXPECT_EQ(length, 42);
 }
 
+TEST_F(EnvTest, CallsValidArenaRef) {
+  Local<String> str = env().NewStringUtf("Foo");
+  ArenaRef arena_ref(env(), str);
+
+  Local<Class> clazz = env().FindClass("java/lang/String");
+  jmethodID to_lower_case =
+      env().GetMethodId(clazz, "toLowerCase", "()Ljava/lang/String;");
+
+  Local<Object> result = env().Call(arena_ref, Method<Object>(to_lower_case));
+  EXPECT_EQ("foo", result.ToString(env()));
+}
+
 TEST_F(EnvTest, GetsStaticFields) {
   Local<Class> clazz = env().FindClass("java/lang/String");
   jfieldID comparator = env().GetStaticFieldId(clazz, "CASE_INSENSITIVE_ORDER",
@@ -160,6 +172,21 @@ TEST_F(EnvTest, ToString) {
   EXPECT_EQ("Foo", result);
 }
 
+TEST_F(EnvTest, IsInstanceOfChecksValidArenaRef) {
+  Local<Class> clazz = env().FindClass("java/lang/String");
+  Local<String> str = env().NewStringUtf("Foo");
+  ArenaRef arena_ref(env(), str);
+  EXPECT_TRUE(env().IsInstanceOf(arena_ref, clazz));
+}
+
+TEST_F(EnvTest, IsInstanceOfChecksNullArenaRef) {
+  GTEST_SKIP()
+      << "Skip until edge case of IsInstanceOf (b/268420201) is covered.";
+  Local<Class> clazz = env().FindClass("java/lang/String");
+  ArenaRef arena_ref;
+  EXPECT_FALSE(env().IsInstanceOf(arena_ref, clazz));
+}
+
 TEST_F(EnvTest, Throw) {
   Local<Class> clazz = env().FindClass("java/lang/Exception");
   jmethodID ctor = env().GetMethodId(clazz, "<init>", "(Ljava/lang/String;)V");
diff --git a/firestore/src/android/document_reference_android.cc b/firestore/src/android/document_reference_android.cc
index f62e7f7407..c5d5176e44 100644
--- a/firestore/src/android/document_reference_android.cc
+++ b/firestore/src/android/document_reference_android.cc
@@ -147,14 +147,14 @@ Future<void> DocumentReferenceInternal::Set(const MapFieldValue& data,
   Env env = GetEnv();
   FieldValueInternal map_value(data);
   Local<Object> java_options = SetOptionsInternal::Create(env, options);
-  Local<Task> task = env.Call(obj_, kSet, map_value, java_options);
+  Local<Task> task = env.Call(obj_, kSet, map_value.ToJava(), java_options);
   return promises_.NewFuture<void>(env, AsyncFn::kSet, task);
 }
 
 Future<void> DocumentReferenceInternal::Update(const MapFieldValue& data) {
   Env env = GetEnv();
   FieldValueInternal map_value(data);
-  Local<Task> task = env.Call(obj_, kUpdate, map_value);
+  Local<Task> task = env.Call(obj_, kUpdate, map_value.ToJava());
   return promises_.NewFuture<void>(env, AsyncFn::kUpdate, task);
 }
 
diff --git a/firestore/src/android/field_value_android.cc b/firestore/src/android/field_value_android.cc
index 071188a726..7e22a406a1 100644
--- a/firestore/src/android/field_value_android.cc
+++ b/firestore/src/android/field_value_android.cc
@@ -38,6 +38,7 @@ namespace firebase {
 namespace firestore {
 namespace {
 
+using jni::ArenaRef;
 using jni::Array;
 using jni::ArrayList;
 using jni::Boolean;
@@ -98,39 +99,45 @@ FieldValue FieldValueInternal::Create(Env& env,
 FieldValueInternal::FieldValueInternal() : cached_type_(Type::kNull) {}
 
 FieldValueInternal::FieldValueInternal(const Object& object)
-    : object_(object), cached_type_(Type::kNull) {}
+    : cached_type_(Type::kNull) {
+  Env env = GetEnv();
+  object_ = ArenaRef(env, object);
+}
 
 FieldValueInternal::FieldValueInternal(Type type, const Object& object)
-    : object_(object), cached_type_(type) {}
+    : cached_type_(type) {
+  Env env = GetEnv();
+  object_ = ArenaRef(env, object);
+}
 
 FieldValueInternal::FieldValueInternal(bool value)
     : cached_type_(Type::kBoolean) {
   Env env = GetEnv();
-  object_ = Boolean::Create(env, value);
+  object_ = ArenaRef(env, Boolean::Create(env, value));
 }
 
 FieldValueInternal::FieldValueInternal(int64_t value)
     : cached_type_(Type::kInteger) {
   Env env = GetEnv();
-  object_ = Long::Create(env, value);
+  object_ = ArenaRef(env, Long::Create(env, value));
 }
 
 FieldValueInternal::FieldValueInternal(double value)
     : cached_type_(Type::kDouble) {
   Env env = GetEnv();
-  object_ = Double::Create(env, value);
+  object_ = ArenaRef(env, Double::Create(env, value));
 }
 
 FieldValueInternal::FieldValueInternal(Timestamp value)
     : cached_type_(Type::kTimestamp) {
   Env env = GetEnv();
-  object_ = TimestampInternal::Create(env, value);
+  object_ = ArenaRef(env, TimestampInternal::Create(env, value));
 }
 
 FieldValueInternal::FieldValueInternal(std::string value)
     : cached_type_(Type::kString) {
   Env env = GetEnv();
-  object_ = env.NewStringUtf(value);
+  object_ = ArenaRef(env, env.NewStringUtf(value));
 }
 
 // We do not initialize cached_blob_ with value here as the instance constructed
@@ -140,20 +147,21 @@ FieldValueInternal::FieldValueInternal(std::string value)
 FieldValueInternal::FieldValueInternal(const uint8_t* value, size_t size)
     : cached_type_(Type::kBlob) {
   Env env = GetEnv();
-  object_ = BlobInternal::Create(env, value, size);
+  object_ = ArenaRef(env, BlobInternal::Create(env, value, size));
 }
 
 FieldValueInternal::FieldValueInternal(DocumentReference value)
     : cached_type_{Type::kReference} {
   if (value.internal_ != nullptr) {
-    object_ = value.internal_->ToJava();
+    Env env = GetEnv();
+    object_ = ArenaRef(env, value.internal_->ToJava());
   }
 }
 
 FieldValueInternal::FieldValueInternal(GeoPoint value)
     : cached_type_(Type::kGeoPoint) {
   Env env = GetEnv();
-  object_ = GeoPointInternal::Create(env, value);
+  object_ = ArenaRef(env, GeoPointInternal::Create(env, value));
 }
 
 FieldValueInternal::FieldValueInternal(const std::vector<FieldValue>& value)
@@ -164,7 +172,7 @@ FieldValueInternal::FieldValueInternal(const std::vector<FieldValue>& value)
     // TODO(b/150016438): don't conflate invalid `FieldValue`s and null.
     list.Add(env, ToJava(element));
   }
-  object_ = list;
+  object_ = ArenaRef(env, list);
 }
 
 FieldValueInternal::FieldValueInternal(const MapFieldValue& value)
@@ -176,20 +184,21 @@ FieldValueInternal::FieldValueInternal(const MapFieldValue& value)
     Local<String> key = env.NewStringUtf(kv.first);
     map.Put(env, key, ToJava(kv.second));
   }
-  object_ = map;
+  object_ = ArenaRef(env, map);
 }
 
 Type FieldValueInternal::type() const {
   if (cached_type_ != Type::kNull) {
     return cached_type_;
   }
-  if (!object_) {
+
+  Env env = GetEnv();
+  if (!object_.get(env)) {
     return Type::kNull;
   }
 
   // We do not have any knowledge on the type yet. Check the runtime type with
   // each known type.
-  Env env = GetEnv();
   if (env.IsInstanceOf(object_, Boolean::GetClass())) {
     cached_type_ = Type::kBoolean;
     return Type::kBoolean;
@@ -232,7 +241,7 @@ Type FieldValueInternal::type() const {
   }
 
   FIREBASE_ASSERT_MESSAGE(false, "Unsupported FieldValue type: %s",
-                          Class::GetClassName(env, object_).c_str());
+                          Class::GetClassName(env, object_.get(env)).c_str());
   return Type::kNull;
 }
 
@@ -389,15 +398,25 @@ bool operator==(const FieldValueInternal& lhs, const FieldValueInternal& rhs) {
 }
 
 template <typename T>
-T FieldValueInternal::Cast(jni::Env& env, Type type) const {
+Local<T> FieldValueInternal::Cast(jni::Env& env, Type type) const {
   if (cached_type_ == Type::kNull) {
     FIREBASE_ASSERT(env.IsInstanceOf(object_, T::GetClass()));
     cached_type_ = type;
   } else {
     FIREBASE_ASSERT(cached_type_ == type);
   }
-  auto typed_value = static_cast<jni::JniType<T>>(object_.get());
-  return T(typed_value);
+  return object_.get(env).CastTo<T>();
+}
+
+template <>
+Local<String> FieldValueInternal::Cast<String>(jni::Env& env, Type type) const {
+  if (cached_type_ == Type::kNull) {
+    FIREBASE_ASSERT(env.IsInstanceOf(object_, String::GetClass()));
+    cached_type_ = type;
+  } else {
+    FIREBASE_ASSERT(cached_type_ == type);
+  }
+  return env.NewStringUtf(object_.get(env).ToString(env));
 }
 
 Local<Array<Object>> FieldValueInternal::MakeArray(
@@ -411,8 +430,14 @@ Local<Array<Object>> FieldValueInternal::MakeArray(
 
 Env FieldValueInternal::GetEnv() { return FirestoreInternal::GetEnv(); }
 
-Object FieldValueInternal::ToJava(const FieldValue& value) {
-  return value.internal_ ? value.internal_->object_ : Object();
+Local<Object> FieldValueInternal::ToJava() const {
+  Env env = GetEnv();
+  return object_.get(env);
+}
+
+Local<Object> FieldValueInternal::ToJava(const FieldValue& value) {
+  Env env = GetEnv();
+  return value.internal_ ? value.internal_->object_.get(env) : Local<Object>();
 }
 
 }  // namespace firestore
diff --git a/firestore/src/android/field_value_android.h b/firestore/src/android/field_value_android.h
index ceb49d23da..6639258413 100644
--- a/firestore/src/android/field_value_android.h
+++ b/firestore/src/android/field_value_android.h
@@ -20,11 +20,14 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <vector>
 
 #include "firebase/firestore/geo_point.h"
 #include "firebase/firestore/timestamp.h"
 #include "firestore/src/include/firebase/firestore/document_reference.h"
 #include "firestore/src/include/firebase/firestore/field_value.h"
+#include "firestore/src/jni/arena_ref.h"
+#include "firestore/src/jni/env.h"
 #include "firestore/src/jni/jni_fwd.h"
 #include "firestore/src/jni/object.h"
 #include "firestore/src/jni/ownership.h"
@@ -87,7 +90,7 @@ class FieldValueInternal {
   std::vector<FieldValue> array_value() const;
   MapFieldValue map_value() const;
 
-  const jni::Global<jni::Object>& ToJava() const { return object_; }
+  jni::Local<jni::Object> ToJava() const;
 
   static FieldValue Delete();
   static FieldValue ServerTimestamp();
@@ -96,7 +99,7 @@ class FieldValueInternal {
   static FieldValue IntegerIncrement(int64_t by_value);
   static FieldValue DoubleIncrement(double by_value);
 
-  static jni::Object ToJava(const FieldValue& value);
+  static jni::Local<jni::Object> ToJava(const FieldValue& value);
 
  private:
   friend class FirestoreInternal;
@@ -107,7 +110,10 @@ class FieldValueInternal {
   // This performs a run-time `instanceof` check to verify that the object
   // has the type `T::GetClass()`.
   template <typename T>
-  T Cast(jni::Env& env, Type type) const;
+  jni::Local<T> Cast(jni::Env& env, Type type) const;
+
+  template <>
+  jni::Local<jni::String> Cast<jni::String>(jni::Env& env, Type type) const;
 
   static jni::Local<jni::Array<jni::Object>> MakeArray(
       jni::Env& env, const std::vector<FieldValue>& elements);
@@ -116,7 +122,7 @@ class FieldValueInternal {
 
   static jni::Env GetEnv();
 
-  jni::Global<jni::Object> object_;
+  jni::ArenaRef object_;
 
   // Below are cached type information. It is costly to get type info from
   // jobject of Object type. So we cache it if we have already known.
@@ -124,19 +130,6 @@ class FieldValueInternal {
   mutable std::shared_ptr<std::vector<uint8_t>> cached_blob_;
 };
 
-inline jni::Object ToJava(const FieldValue& value) {
-  // This indirection is required to make use of the `friend` in FieldValue.
-  return FieldValueInternal::ToJava(value);
-}
-
-inline jobject ToJni(const FieldValueInternal* value) {
-  return value->ToJava().get();
-}
-
-inline jobject ToJni(const FieldValueInternal& value) {
-  return value.ToJava().get();
-}
-
 }  // namespace firestore
 }  // namespace firebase
 
diff --git a/firestore/src/android/firestore_android.cc b/firestore/src/android/firestore_android.cc
index 62917accd8..3d41b7cb87 100644
--- a/firestore/src/android/firestore_android.cc
+++ b/firestore/src/android/firestore_android.cc
@@ -59,6 +59,7 @@
 #include "firestore/src/common/hard_assert_common.h"
 #include "firestore/src/common/make_unique.h"
 #include "firestore/src/include/firebase/firestore.h"
+#include "firestore/src/jni/arena_ref.h"
 #include "firestore/src/jni/array.h"
 #include "firestore/src/jni/array_list.h"
 #include "firestore/src/jni/boolean.h"
@@ -314,6 +315,10 @@ bool FirestoreInternal::Initialize(App* app) {
     jni::Long::Initialize(loader);
     jni::Map::Initialize(loader);
 
+    // Initialize ArenaRef _after_ the other JNI classes because it relies on
+    // HashMap having already been initialized.
+    jni::ArenaRef::Initialize(env);
+
     InitializeFirestore(loader);
     InitializeFirestoreTasks(loader);
     InitializeUserCallbackExecutor(loader);
diff --git a/firestore/src/android/query_android.cc b/firestore/src/android/query_android.cc
index b28de7eb81..95953c0383 100644
--- a/firestore/src/android/query_android.cc
+++ b/firestore/src/android/query_android.cc
@@ -272,7 +272,8 @@ Query QueryInternal::Where(const FieldPath& field,
                            const FieldValue& value) const {
   Env env = GetEnv();
   Local<Object> java_field = FieldPathConverter::Create(env, field);
-  Local<Object> query = env.Call(obj_, method, java_field, ToJava(value));
+  Local<Object> java_value = FieldValueInternal::ToJava(value);
+  Local<Object> query = env.Call(obj_, method, java_field, java_value);
   return firestore_->NewQuery(env, query);
 }
 
@@ -284,7 +285,7 @@ Query QueryInternal::Where(const FieldPath& field,
   size_t size = values.size();
   Local<ArrayList> java_values = ArrayList::Create(env, size);
   for (size_t i = 0; i < size; ++i) {
-    java_values.Add(env, ToJava(values[i]));
+    java_values.Add(env, FieldValueInternal::ToJava(values[i]));
   }
 
   Local<Object> java_field = FieldPathConverter::Create(env, field);
@@ -328,9 +329,9 @@ ListenerRegistration QueryInternal::AddSnapshotListener(
   Local<Object> java_metadata =
       MetadataChangesInternal::Create(env, metadata_changes);
 
-  Local<Object> java_registration =
-      env.Call(obj_, kAddSnapshotListener, firestore_->user_callback_executor(),
-               java_metadata, java_listener);
+  Local<Object> java_registration = env.Call(
+      obj_.get(env), kAddSnapshotListener, firestore_->user_callback_executor(),
+      java_metadata, java_listener);
 
   if (!env.ok()) return {};
   return ListenerRegistration(new ListenerRegistrationInternal(
@@ -342,7 +343,7 @@ Local<Array<Object>> QueryInternal::ConvertFieldValues(
   size_t size = field_values.size();
   Local<Array<Object>> result = env.NewArray(size, Object::GetClass());
   for (size_t i = 0; i < size; ++i) {
-    result.Set(env, i, ToJava(field_values[i]));
+    result.Set(env, i, FieldValueInternal::ToJava(field_values[i]));
   }
   return result;
 }
diff --git a/firestore/src/android/util_android.cc b/firestore/src/android/util_android.cc
index 5e9a534499..f7b6126caf 100644
--- a/firestore/src/android/util_android.cc
+++ b/firestore/src/android/util_android.cc
@@ -36,7 +36,7 @@ Local<HashMap> MakeJavaMap(Env& env, const MapFieldValue& data) {
   Local<HashMap> result = HashMap::Create(env);
   for (const auto& kv : data) {
     Local<String> key = env.NewStringUtf(kv.first);
-    const Object& value = ToJava(kv.second);
+    Local<Object> value = FieldValueInternal::ToJava(kv.second);
     result.Put(env, key, value);
   }
   return result;
@@ -49,7 +49,7 @@ UpdateFieldPathArgs MakeUpdateFieldPathArgs(Env& env,
   FIREBASE_DEV_ASSERT_MESSAGE(iter != end, "data must be non-empty");
 
   Local<Object> first_field = FieldPathConverter::Create(env, iter->first);
-  const Object& first_value = ToJava(iter->second);
+  Local<Object> first_value = FieldValueInternal::ToJava(iter->second);
   ++iter;
 
   const auto size = std::distance(iter, end) * 2;
@@ -58,13 +58,14 @@ UpdateFieldPathArgs MakeUpdateFieldPathArgs(Env& env,
   int index = 0;
   for (; iter != end; ++iter) {
     Local<Object> field = FieldPathConverter::Create(env, iter->first);
-    const Object& value = ToJava(iter->second);
+    Local<Object> value = FieldValueInternal::ToJava(iter->second);
 
     varargs.Set(env, index++, field);
     varargs.Set(env, index++, value);
   }
 
-  return UpdateFieldPathArgs{Move(first_field), first_value, Move(varargs)};
+  return UpdateFieldPathArgs{Move(first_field), Move(first_value),
+                             Move(varargs)};
 }
 
 }  // namespace firestore
diff --git a/firestore/src/android/util_android.h b/firestore/src/android/util_android.h
index c0e43c0ae0..2dcd03d32b 100644
--- a/firestore/src/android/util_android.h
+++ b/firestore/src/android/util_android.h
@@ -70,7 +70,7 @@ jni::Local<jni::HashMap> MakeJavaMap(jni::Env& env, const MapFieldValue& data);
  */
 struct UpdateFieldPathArgs {
   jni::Local<jni::Object> first_field;
-  jni::Object first_value;
+  jni::Local<jni::Object> first_value;
   jni::Local<jni::Array<jni::Object>> varargs;
 };
 
diff --git a/firestore/src/android/wrapper.cc b/firestore/src/android/wrapper.cc
index c4304abe29..29a294fcbd 100644
--- a/firestore/src/android/wrapper.cc
+++ b/firestore/src/android/wrapper.cc
@@ -16,8 +16,6 @@
 
 #include "firestore/src/android/wrapper.h"
 
-#include <iterator>
-
 #include "app/src/assert.h"
 #include "firestore/src/android/field_path_android.h"
 #include "firestore/src/android/field_value_android.h"
@@ -29,14 +27,18 @@ namespace firebase {
 namespace firestore {
 namespace {
 
+using jni::ArenaRef;
 using jni::Env;
+using jni::Local;
 using jni::Object;
 
 }  // namespace
 
 Wrapper::Wrapper(FirestoreInternal* firestore, const Object& obj)
-    : firestore_(firestore), obj_(obj) {
+    : firestore_(firestore) {
   FIREBASE_ASSERT(obj);
+  Env env = GetEnv();
+  obj_ = ArenaRef(env, obj);
 }
 
 Wrapper::Wrapper() {
@@ -56,11 +58,12 @@ Wrapper::Wrapper(Wrapper* rhs) : Wrapper() {
 
 Wrapper::~Wrapper() = default;
 
-jni::Env Wrapper::GetEnv() const { return firestore_->GetEnv(); }
-
-Object Wrapper::ToJava(const FieldValue& value) {
-  return FieldValueInternal::ToJava(value);
+Local<Object> Wrapper::ToJava() const {
+  Env env = GetEnv();
+  return obj_.get(env);
 }
 
+jni::Env Wrapper::GetEnv() const { return firestore_->GetEnv(); }
+
 }  // namespace firestore
 }  // namespace firebase
diff --git a/firestore/src/android/wrapper.h b/firestore/src/android/wrapper.h
index b1da5fce65..d88a1baed1 100644
--- a/firestore/src/android/wrapper.h
+++ b/firestore/src/android/wrapper.h
@@ -17,6 +17,8 @@
 #ifndef FIREBASE_FIRESTORE_SRC_ANDROID_WRAPPER_H_
 #define FIREBASE_FIRESTORE_SRC_ANDROID_WRAPPER_H_
 
+#include "firestore/src/jni/arena_ref.h"
+#include "firestore/src/jni/env.h"
 #include "firestore/src/jni/jni_fwd.h"
 #include "firestore/src/jni/object.h"
 #include "firestore/src/jni/ownership.h"
@@ -44,9 +46,7 @@ class Wrapper {
   Wrapper& operator=(Wrapper&& wrapper) = delete;
 
   FirestoreInternal* firestore_internal() { return firestore_; }
-  const jni::Object& ToJava() const { return obj_; }
-
-  static jni::Object ToJava(const FieldValue& value);
+  jni::Local<jni::Object> ToJava() const;
 
  protected:
   // Default constructor. Subclass is expected to set the obj_ a meaningful
@@ -59,7 +59,7 @@ class Wrapper {
   jni::Env GetEnv() const;
 
   FirestoreInternal* firestore_ = nullptr;  // not owning
-  jni::Global<jni::Object> obj_;
+  jni::ArenaRef obj_;
 
  private:
   friend class FirestoreInternal;
diff --git a/firestore/src/android/write_batch_android.cc b/firestore/src/android/write_batch_android.cc
index 31b659308f..35b9367df8 100644
--- a/firestore/src/android/write_batch_android.cc
+++ b/firestore/src/android/write_batch_android.cc
@@ -107,8 +107,8 @@ Future<void> WriteBatchInternal::Commit() {
   return promises_.NewFuture<void>(env, AsyncFn::kCommit, task);
 }
 
-jni::Object WriteBatchInternal::ToJava(const DocumentReference& reference) {
-  return reference.internal_ ? reference.internal_->ToJava() : jni::Object();
+Local<Object> WriteBatchInternal::ToJava(const DocumentReference& reference) {
+  return reference.internal_ ? reference.internal_->ToJava() : Local<Object>();
 }
 
 }  // namespace firestore
diff --git a/firestore/src/android/write_batch_android.h b/firestore/src/android/write_batch_android.h
index 86c7247bbc..e309ae3d78 100644
--- a/firestore/src/android/write_batch_android.h
+++ b/firestore/src/android/write_batch_android.h
@@ -64,7 +64,7 @@ class WriteBatchInternal : public Wrapper {
    */
   // TODO(mcg): Move this out of WriteBatchInternal
   // This needs to be here now because of existing friend relationships.
-  static jni::Object ToJava(const DocumentReference& reference);
+  static jni::Local<jni::Object> ToJava(const DocumentReference& reference);
 
   PromiseFactory<AsyncFn> promises_;
 };
diff --git a/firestore/src/jni/arena_ref.cc b/firestore/src/jni/arena_ref.cc
new file mode 100644
index 0000000000..38b66b67a5
--- /dev/null
+++ b/firestore/src/jni/arena_ref.cc
@@ -0,0 +1,146 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * 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.
+ */
+
+#include "firestore/src/jni/arena_ref.h"
+
+#include <atomic>
+#include <utility>
+
+#include "firestore/src/android/firestore_android.h"
+#include "firestore/src/jni/env.h"
+#include "firestore/src/jni/hash_map.h"
+#include "firestore/src/jni/long.h"
+#include "firestore/src/jni/map.h"
+
+namespace firebase {
+namespace firestore {
+namespace jni {
+
+namespace {
+
+HashMap* gArenaRefHashMap = nullptr;
+jclass gLongClass = nullptr;
+jmethodID gLongConstructor = nullptr;
+std::mutex mutex_;
+
+int64_t GetNextArenaRefKey() {
+  static std::atomic<int64_t> next_key(0);
+  return next_key.fetch_add(1);
+}
+
+}  // namespace
+
+ArenaRef::ArenaRef(Env& env, const Object& object)
+    : key_(GetNextArenaRefKey()) {
+  Local<Long> key_ref = key_object(env);
+  std::lock_guard<std::mutex> lock(mutex_);
+  gArenaRefHashMap->Put(env, key_ref, object);
+}
+
+ArenaRef::ArenaRef(const ArenaRef& other)
+    : key_(other.key_ == kInvalidKey ? kInvalidKey : GetNextArenaRefKey()) {
+  if (other.key_ != kInvalidKey) {
+    Env env;
+    Local<Object> object = other.get(env);
+    Local<Long> key_ref = key_object(env);
+    std::lock_guard<std::mutex> lock(mutex_);
+    gArenaRefHashMap->Put(env, key_ref, object);
+  }
+}
+
+ArenaRef::ArenaRef(ArenaRef&& other) {
+  using std::swap;  // go/using-std-swap
+  swap(key_, other.key_);
+}
+
+ArenaRef& ArenaRef::operator=(const ArenaRef& other) {
+  if (this == &other || (key_ == kInvalidKey && other.key_ == kInvalidKey)) {
+    return *this;
+  }
+
+  Env env;
+  if (other.key_ != kInvalidKey) {
+    if (key_ == kInvalidKey) {
+      key_ = GetNextArenaRefKey();
+    }
+    Local<Object> object = other.get(env);
+    std::lock_guard<std::mutex> lock(mutex_);
+    gArenaRefHashMap->Put(env, key_object(env), object);
+  } else {
+    {
+      std::lock_guard<std::mutex> lock(mutex_);
+      gArenaRefHashMap->Remove(env, key_object(env));
+    }
+    key_ = kInvalidKey;
+  }
+  return *this;
+}
+
+ArenaRef& ArenaRef::operator=(ArenaRef&& other) {
+  if (this != &other) {
+    std::swap(key_, other.key_);
+  }
+  return *this;
+}
+
+ArenaRef::~ArenaRef() {
+  if (key_ != kInvalidKey) {
+    Env env;
+    ExceptionClearGuard block(env);
+    {
+      std::lock_guard<std::mutex> lock(mutex_);
+      gArenaRefHashMap->Remove(env, key_object(env));
+    }
+    if (!env.ok()) {
+      env.RecordException();
+      env.ExceptionClear();
+    }
+  }
+}
+
+Local<Long> ArenaRef::key_object(Env& env) const {
+  if (!env.ok()) return {};
+  // The reason why using raw JNI calls here rather than the JNI convenience
+  // layer is because we need to get rid of JNI convenience layer dependencies
+  // in the ArenaRef destructor to avoid racing condition when firestore object
+  // gets destroyed.
+  jobject key = env.get()->NewObject(gLongClass, gLongConstructor, key_);
+  env.RecordException();
+  return {env.get(), key};
+}
+
+void ArenaRef::Initialize(Env& env) {
+  if (gArenaRefHashMap) {
+    return;
+  }
+  Global<HashMap> hash_map(HashMap::Create(env));
+  gArenaRefHashMap = new HashMap(hash_map.release());
+
+  auto long_class_local = env.get()->FindClass("java/lang/Long");
+  gLongClass = static_cast<jclass>(env.get()->NewGlobalRef(long_class_local));
+  env.get()->DeleteLocalRef(long_class_local);
+  gLongConstructor = env.get()->GetMethodID(gLongClass, "<init>", "(J)V");
+
+  if (!env.ok()) return;
+}
+
+Local<Object> ArenaRef::get(Env& env) const {
+  return gArenaRefHashMap->Get(env, key_object(env));
+}
+
+}  // namespace jni
+}  // namespace firestore
+}  // namespace firebase
diff --git a/firestore/src/jni/arena_ref.h b/firestore/src/jni/arena_ref.h
new file mode 100644
index 0000000000..8ccbb43fab
--- /dev/null
+++ b/firestore/src/jni/arena_ref.h
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * 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.
+ */
+
+#ifndef FIREBASE_FIRESTORE_SRC_JNI_ARENA_REF_H_
+#define FIREBASE_FIRESTORE_SRC_JNI_ARENA_REF_H_
+
+#include <jni.h>
+
+#include <cstdlib>
+
+#include "firestore/src/jni/jni_fwd.h"
+#include "firestore/src/jni/object.h"
+#include "firestore/src/jni/ownership.h"
+
+namespace firebase {
+namespace firestore {
+namespace jni {
+
+/**
+ * ArenaRef is an RAII wrapper which serves the same purpose as Global<T>, both
+ * of them manage the lifetime of JNI reference. Compared to Global<T>, the
+ * advantage of ArenaRef is it gets rid of the restriction of total number of
+ * JNI global reference (the pointer inside Global<T>) need to be under 51,200,
+ * since after that number the app will crash. The drawback is ArenaRef holds
+ * JNI Reference indirectly using HashMap so there is a efficiency lost against
+ * Global<T>.
+ */
+class ArenaRef {
+ public:
+  ArenaRef() = default;
+  ArenaRef(Env&, const Object&);
+  ArenaRef(const ArenaRef& other);
+  ArenaRef(ArenaRef&& other);
+  ArenaRef& operator=(const ArenaRef& other);
+  ArenaRef& operator=(ArenaRef&& other);
+  ~ArenaRef();
+
+  static void Initialize(Env&);
+
+  Local<Object> get(Env&) const;
+
+ private:
+  Local<Long> key_object(Env&) const;
+
+  static constexpr int64_t kInvalidKey = -1;
+  int64_t key_ = kInvalidKey;
+};
+
+}  // namespace jni
+}  // namespace firestore
+}  // namespace firebase
+
+#endif  // FIREBASE_FIRESTORE_SRC_JNI_ARENA_REF_H_
diff --git a/firestore/src/jni/env.h b/firestore/src/jni/env.h
index a7c3354d68..3906c35d83 100644
--- a/firestore/src/jni/env.h
+++ b/firestore/src/jni/env.h
@@ -24,6 +24,7 @@
 #include <vector>
 
 #include "app/meta/move.h"
+#include "firestore/src/jni/arena_ref.h"
 #include "firestore/src/jni/call_traits.h"
 #include "firestore/src/jni/class.h"
 #include "firestore/src/jni/declaration.h"
@@ -122,6 +123,9 @@ class Env {
   /** Clears the last exception. */
   void ExceptionClear();
 
+  /** Prints out exception to logcat if `Env` has an exception. */
+  void RecordException();
+
   /**
    * Returns the last Java exception to occur and clears the pending exception.
    */
@@ -183,6 +187,9 @@ class Env {
   bool IsInstanceOf(jobject object, jclass clazz) {
     return IsInstanceOf(Object(object), Class(clazz));
   }
+  bool IsInstanceOf(const ArenaRef& object, const Class& clazz) {
+    return IsInstanceOf(object.get(*this), Class(clazz));
+  }
 
   bool IsSameObject(const Object& object1, const Object& object2);
 
@@ -240,6 +247,13 @@ class Env {
                          ToJni(Forward<Args>(args))...);
   }
 
+  template <typename T, typename... Args>
+  ResultType<T> Call(const ArenaRef& object,
+                     const Method<T>& method,
+                     Args&&... args) {
+    return Call(object.get(*this), method, Forward<Args>(args)...);
+  }
+
   // MARK: Accessing Static Fields
 
   jfieldID GetStaticFieldId(const Class& clazz,
@@ -513,7 +527,6 @@ class Env {
     RecordException();
   }
 
-  void RecordException();
   std::string ErrorDescription(const Object& object);
   const char* ErrorName(jint error);
 
diff --git a/firestore/src/jni/jni_fwd.h b/firestore/src/jni/jni_fwd.h
index 5969ddc2c1..a2f7e0aff0 100644
--- a/firestore/src/jni/jni_fwd.h
+++ b/firestore/src/jni/jni_fwd.h
@@ -38,6 +38,7 @@ template <typename T>
 class Global;
 template <typename T>
 class NonOwning;
+class ArenaRef;
 
 class Class;
 class Object;