andreacavalli/rocksdb
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add event listeners to RocksJava (#7425)

Browse Source 
Summary:
Allows adding event listeners in RocksJava.

* Adds listeners getter and setter in `Options` and `DBOptions` classes.
* Adds `EventListener` Java interface and base class for implementing custom event listener callbacks - `AbstractEventListener`, which has an underlying native callback class implementing C++ `EventListener` class.
* `AbstractEventListener` class has mechanism for selectively enabling its callback methods in order to prevent invoking Java method if it is not implemented. This decreases performance cost in case only subset of event listener callback methods is needed - the JNI code for remaining "no-op" callbacks is not executed.
* The code is covered by unit tests in `EventListenerTest.java`, there are also tests added for setting/getting listeners field in `OptionsTest.java` and `DBOptionsTest.java`.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/7425

Reviewed By: pdillinger

Differential Revision: D24063390

Pulled By: jay-zhuang

fbshipit-source-id: 508c359538983d6b765e70d9989c351794a944ee
This commit is contained in:
Tomasz Posluszny 2020-10-14 11:32:10 -07:00 committed by Facebook GitHub Bot
parent b99fe1ab74
commit 6528ecc800
36 changed files with 4324 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
java/CMakeLists.txt
View File

@ -30,6 +30,8 @@ set(JNI_NATIVE_SOURCES
rocksjni/config_options.cc
rocksjni/env.cc
rocksjni/env_options.cc
rocksjni/event_listener.cc
rocksjni/event_listener_jnicallback.cc
rocksjni/filter.cc
rocksjni/ingest_external_file_options.cc
rocksjni/iterator.cc
@ -87,6 +89,7 @@ set(JAVA_MAIN_CLASSES
src/main/java/org/rocksdb/AbstractCompactionFilter.java
src/main/java/org/rocksdb/AbstractCompactionFilterFactory.java
src/main/java/org/rocksdb/AbstractComparator.java
src/main/java/org/rocksdb/AbstractEventListener.java
src/main/java/org/rocksdb/AbstractImmutableNativeReference.java
src/main/java/org/rocksdb/AbstractMutableOptions.java
src/main/java/org/rocksdb/AbstractNativeReference.java
@ -100,6 +103,7 @@ set(JAVA_MAIN_CLASSES
src/main/java/org/rocksdb/AccessHint.java
src/main/java/org/rocksdb/AdvancedColumnFamilyOptionsInterface.java
src/main/java/org/rocksdb/AdvancedMutableColumnFamilyOptionsInterface.java
src/main/java/org/rocksdb/BackgroundErrorReason.java
src/main/java/org/rocksdb/BackupableDBOptions.java
src/main/java/org/rocksdb/BackupEngine.java
src/main/java/org/rocksdb/BackupInfo.java
@ -140,8 +144,13 @@ set(JAVA_MAIN_CLASSES
src/main/java/org/rocksdb/EncodingType.java
src/main/java/org/rocksdb/Env.java
src/main/java/org/rocksdb/EnvOptions.java
src/main/java/org/rocksdb/EventListener.java
src/main/java/org/rocksdb/Experimental.java
src/main/java/org/rocksdb/ExternalFileIngestionInfo.java
src/main/java/org/rocksdb/Filter.java
src/main/java/org/rocksdb/FileOperationInfo.java
src/main/java/org/rocksdb/FlushJobInfo.java
src/main/java/org/rocksdb/FlushReason.java
src/main/java/org/rocksdb/FlushOptions.java
src/main/java/org/rocksdb/HashLinkedListMemTableConfig.java
src/main/java/org/rocksdb/HashSkipListMemTableConfig.java
@ -163,6 +172,7 @@ set(JAVA_MAIN_CLASSES
src/main/java/org/rocksdb/MemoryUsageType.java
src/main/java/org/rocksdb/MemoryUtil.java
src/main/java/org/rocksdb/MemTableConfig.java
src/main/java/org/rocksdb/MemTableInfo.java
src/main/java/org/rocksdb/MergeOperator.java
src/main/java/org/rocksdb/MutableColumnFamilyOptions.java
src/main/java/org/rocksdb/MutableColumnFamilyOptionsInterface.java
@ -218,6 +228,10 @@ set(JAVA_MAIN_CLASSES
src/main/java/org/rocksdb/StatsLevel.java
src/main/java/org/rocksdb/Status.java
src/main/java/org/rocksdb/StringAppendOperator.java
src/main/java/org/rocksdb/TableFileCreationBriefInfo.java
src/main/java/org/rocksdb/TableFileCreationInfo.java
src/main/java/org/rocksdb/TableFileCreationReason.java
src/main/java/org/rocksdb/TableFileDeletionInfo.java
src/main/java/org/rocksdb/TableFilter.java
src/main/java/org/rocksdb/TableProperties.java
src/main/java/org/rocksdb/TableFormatConfig.java
@ -247,6 +261,8 @@ set(JAVA_MAIN_CLASSES
src/main/java/org/rocksdb/WriteBatchWithIndex.java
src/main/java/org/rocksdb/WriteOptions.java
src/main/java/org/rocksdb/WriteBufferManager.java
src/main/java/org/rocksdb/WriteStallCondition.java
src/main/java/org/rocksdb/WriteStallInfo.java
src/main/java/org/rocksdb/util/ByteUtil.java
src/main/java/org/rocksdb/util/BytewiseComparator.java
src/main/java/org/rocksdb/util/Environment.java
@ -391,6 +407,7 @@ if(${CMAKE_VERSION} VERSION_LESS "3.11.4" OR (${Java_VERSION_MINOR} STREQUAL "7"
org.rocksdb.AbstractCompactionFilter
org.rocksdb.AbstractCompactionFilterFactory
org.rocksdb.AbstractComparator
org.rocksdb.AbstractEventListener
org.rocksdb.AbstractImmutableNativeReference
org.rocksdb.AbstractNativeReference
org.rocksdb.AbstractRocksIterator

2
java/Makefile
View File

@ -2,6 +2,7 @@ NATIVE_JAVA_CLASSES = \
org.rocksdb.AbstractCompactionFilter\
org.rocksdb.AbstractCompactionFilterFactory\
org.rocksdb.AbstractComparator\
org.rocksdb.AbstractEventListener\
org.rocksdb.AbstractSlice\
org.rocksdb.AbstractTableFilter\
org.rocksdb.AbstractTraceWriter\
@ -129,6 +130,7 @@ JAVA_TESTS = \
org.rocksdb.DirectSliceTest\
org.rocksdb.util.EnvironmentTest\
org.rocksdb.EnvOptionsTest\
org.rocksdb.EventListenerTest\
org.rocksdb.HdfsEnvTest\
org.rocksdb.IngestExternalFileOptionsTest\
org.rocksdb.util.IntComparatorTest\

43
java/rocksjni/event_listener.cc Normal file
View File

@ -0,0 +1,43 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
//
// This file implements the "bridge" between Java and C++ for
// rocksdb::EventListener.
#include <jni.h>
#include <memory>
#include "include/org_rocksdb_AbstractEventListener.h"
#include "rocksjni/event_listener_jnicallback.h"
#include "rocksjni/portal.h"
/*
* Class: org_rocksdb_AbstractEventListener
* Method: createNewEventListener
* Signature: (J)J
*/
jlong Java_org_rocksdb_AbstractEventListener_createNewEventListener(
JNIEnv* env, jobject jobj, jlong jenabled_event_callback_values) {
auto enabled_event_callbacks =
ROCKSDB_NAMESPACE::EnabledEventCallbackJni::toCppEnabledEventCallbacks(
jenabled_event_callback_values);
auto* sptr_event_listener =
new std::shared_ptr<ROCKSDB_NAMESPACE::EventListener>(
new ROCKSDB_NAMESPACE::EventListenerJniCallback(
env, jobj, enabled_event_callbacks));
return reinterpret_cast<jlong>(sptr_event_listener);
}
/*
* Class: org_rocksdb_AbstractEventListener
* Method: disposeInternal
* Signature: (J)V
*/
void Java_org_rocksdb_AbstractEventListener_disposeInternal(JNIEnv*, jobject,
jlong jhandle) {
delete reinterpret_cast<std::shared_ptr<ROCKSDB_NAMESPACE::EventListener>*>(
jhandle);
}

502
java/rocksjni/event_listener_jnicallback.cc Normal file
View File

@ -0,0 +1,502 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
//
// This file implements the callback "bridge" between Java and C++ for
// rocksdb::EventListener.
#include "rocksjni/event_listener_jnicallback.h"
#include "rocksjni/portal.h"
namespace rocksdb {
EventListenerJniCallback::EventListenerJniCallback(
JNIEnv* env, jobject jevent_listener,
const std::set<EnabledEventCallback>& enabled_event_callbacks)
: JniCallback(env, jevent_listener),
m_enabled_event_callbacks(enabled_event_callbacks) {
InitCallbackMethodId(
m_on_flush_completed_proxy_mid, EnabledEventCallback::ON_FLUSH_COMPLETED,
env, AbstractEventListenerJni::getOnFlushCompletedProxyMethodId);
InitCallbackMethodId(m_on_flush_begin_proxy_mid,
EnabledEventCallback::ON_FLUSH_BEGIN, env,
AbstractEventListenerJni::getOnFlushBeginProxyMethodId);
InitCallbackMethodId(m_on_table_file_deleted_mid,
EnabledEventCallback::ON_TABLE_FILE_DELETED, env,
AbstractEventListenerJni::getOnTableFileDeletedMethodId);
InitCallbackMethodId(
m_on_compaction_begin_proxy_mid,
EnabledEventCallback::ON_COMPACTION_BEGIN, env,
AbstractEventListenerJni::getOnCompactionBeginProxyMethodId);
InitCallbackMethodId(
m_on_compaction_completed_proxy_mid,
EnabledEventCallback::ON_COMPACTION_COMPLETED, env,
AbstractEventListenerJni::getOnCompactionCompletedProxyMethodId);
InitCallbackMethodId(m_on_table_file_created_mid,
EnabledEventCallback::ON_TABLE_FILE_CREATED, env,
AbstractEventListenerJni::getOnTableFileCreatedMethodId);
InitCallbackMethodId(
m_on_table_file_creation_started_mid,
EnabledEventCallback::ON_TABLE_FILE_CREATION_STARTED, env,
AbstractEventListenerJni::getOnTableFileCreationStartedMethodId);
InitCallbackMethodId(m_on_mem_table_sealed_mid,
EnabledEventCallback::ON_MEMTABLE_SEALED, env,
AbstractEventListenerJni::getOnMemTableSealedMethodId);
InitCallbackMethodId(
m_on_column_family_handle_deletion_started_mid,
EnabledEventCallback::ON_COLUMN_FAMILY_HANDLE_DELETION_STARTED, env,
AbstractEventListenerJni::getOnColumnFamilyHandleDeletionStartedMethodId);
InitCallbackMethodId(
m_on_external_file_ingested_proxy_mid,
EnabledEventCallback::ON_EXTERNAL_FILE_INGESTED, env,
AbstractEventListenerJni::getOnExternalFileIngestedProxyMethodId);
InitCallbackMethodId(
m_on_background_error_proxy_mid,
EnabledEventCallback::ON_BACKGROUND_ERROR, env,
AbstractEventListenerJni::getOnBackgroundErrorProxyMethodId);
InitCallbackMethodId(
m_on_stall_conditions_changed_mid,
EnabledEventCallback::ON_STALL_CONDITIONS_CHANGED, env,
AbstractEventListenerJni::getOnStallConditionsChangedMethodId);
InitCallbackMethodId(m_on_file_read_finish_mid,
EnabledEventCallback::ON_FILE_READ_FINISH, env,
AbstractEventListenerJni::getOnFileReadFinishMethodId);
InitCallbackMethodId(m_on_file_write_finish_mid,
EnabledEventCallback::ON_FILE_WRITE_FINISH, env,
AbstractEventListenerJni::getOnFileWriteFinishMethodId);
InitCallbackMethodId(m_on_file_flush_finish_mid,
EnabledEventCallback::ON_FILE_FLUSH_FINISH, env,
AbstractEventListenerJni::getOnFileFlushFinishMethodId);
InitCallbackMethodId(m_on_file_sync_finish_mid,
EnabledEventCallback::ON_FILE_SYNC_FINISH, env,
AbstractEventListenerJni::getOnFileSyncFinishMethodId);
InitCallbackMethodId(
m_on_file_range_sync_finish_mid,
EnabledEventCallback::ON_FILE_RANGE_SYNC_FINISH, env,
AbstractEventListenerJni::getOnFileRangeSyncFinishMethodId);
InitCallbackMethodId(
m_on_file_truncate_finish_mid,
EnabledEventCallback::ON_FILE_TRUNCATE_FINISH, env,
AbstractEventListenerJni::getOnFileTruncateFinishMethodId);
InitCallbackMethodId(m_on_file_close_finish_mid,
EnabledEventCallback::ON_FILE_CLOSE_FINISH, env,
AbstractEventListenerJni::getOnFileCloseFinishMethodId);
InitCallbackMethodId(
m_should_be_notified_on_file_io,
EnabledEventCallback::SHOULD_BE_NOTIFIED_ON_FILE_IO, env,
AbstractEventListenerJni::getShouldBeNotifiedOnFileIOMethodId);
InitCallbackMethodId(
m_on_error_recovery_begin_proxy_mid,
EnabledEventCallback::ON_ERROR_RECOVERY_BEGIN, env,
AbstractEventListenerJni::getOnErrorRecoveryBeginProxyMethodId);
InitCallbackMethodId(
m_on_error_recovery_completed_mid,
EnabledEventCallback::ON_ERROR_RECOVERY_COMPLETED, env,
AbstractEventListenerJni::getOnErrorRecoveryCompletedMethodId);
}
EventListenerJniCallback::~EventListenerJniCallback() {}
void EventListenerJniCallback::OnFlushCompleted(
DB* db, const FlushJobInfo& flush_job_info) {
if (m_on_flush_completed_proxy_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jflush_job_info = SetupCallbackInvocation<FlushJobInfo>(
env, attached_thread, flush_job_info,
FlushJobInfoJni::fromCppFlushJobInfo);
if (jflush_job_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_flush_completed_proxy_mid,
reinterpret_cast<jlong>(db), jflush_job_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jflush_job_info});
}
void EventListenerJniCallback::OnFlushBegin(
DB* db, const FlushJobInfo& flush_job_info) {
if (m_on_flush_begin_proxy_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jflush_job_info = SetupCallbackInvocation<FlushJobInfo>(
env, attached_thread, flush_job_info,
FlushJobInfoJni::fromCppFlushJobInfo);
if (jflush_job_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_flush_begin_proxy_mid,
reinterpret_cast<jlong>(db), jflush_job_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jflush_job_info});
}
void EventListenerJniCallback::OnTableFileDeleted(
const TableFileDeletionInfo& info) {
if (m_on_table_file_deleted_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jdeletion_info = SetupCallbackInvocation<TableFileDeletionInfo>(
env, attached_thread, info,
TableFileDeletionInfoJni::fromCppTableFileDeletionInfo);
if (jdeletion_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_table_file_deleted_mid,
jdeletion_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jdeletion_info});
}
void EventListenerJniCallback::OnCompactionBegin(DB* db,
const CompactionJobInfo& ci) {
if (m_on_compaction_begin_proxy_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jcompaction_job_info = SetupCallbackInvocation<CompactionJobInfo>(
env, attached_thread, ci, CompactionJobInfoJni::fromCppCompactionJobInfo);
if (jcompaction_job_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_compaction_begin_proxy_mid,
reinterpret_cast<jlong>(db), jcompaction_job_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jcompaction_job_info});
}
void EventListenerJniCallback::OnCompactionCompleted(
DB* db, const CompactionJobInfo& ci) {
if (m_on_compaction_completed_proxy_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jcompaction_job_info = SetupCallbackInvocation<CompactionJobInfo>(
env, attached_thread, ci, CompactionJobInfoJni::fromCppCompactionJobInfo);
if (jcompaction_job_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_compaction_completed_proxy_mid,
reinterpret_cast<jlong>(db), jcompaction_job_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jcompaction_job_info});
}
void EventListenerJniCallback::OnTableFileCreated(
const TableFileCreationInfo& info) {
if (m_on_table_file_created_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jfile_creation_info = SetupCallbackInvocation<TableFileCreationInfo>(
env, attached_thread, info,
TableFileCreationInfoJni::fromCppTableFileCreationInfo);
if (jfile_creation_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_table_file_created_mid,
jfile_creation_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jfile_creation_info});
}
void EventListenerJniCallback::OnTableFileCreationStarted(
const TableFileCreationBriefInfo& info) {
if (m_on_table_file_creation_started_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jcreation_brief_info =
SetupCallbackInvocation<TableFileCreationBriefInfo>(
env, attached_thread, info,
TableFileCreationBriefInfoJni::fromCppTableFileCreationBriefInfo);
if (jcreation_brief_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_table_file_creation_started_mid,
jcreation_brief_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jcreation_brief_info});
}
void EventListenerJniCallback::OnMemTableSealed(const MemTableInfo& info) {
if (m_on_mem_table_sealed_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jmem_table_info = SetupCallbackInvocation<MemTableInfo>(
env, attached_thread, info, MemTableInfoJni::fromCppMemTableInfo);
if (jmem_table_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_mem_table_sealed_mid,
jmem_table_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jmem_table_info});
}
void EventListenerJniCallback::OnColumnFamilyHandleDeletionStarted(
ColumnFamilyHandle* handle) {
if (m_on_column_family_handle_deletion_started_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jcf_handle = SetupCallbackInvocation<ColumnFamilyHandle>(
env, attached_thread, *handle,
ColumnFamilyHandleJni::fromCppColumnFamilyHandle);
if (jcf_handle != nullptr) {
env->CallVoidMethod(m_jcallback_obj,
m_on_column_family_handle_deletion_started_mid,
jcf_handle);
}
CleanupCallbackInvocation(env, attached_thread, {&jcf_handle});
}
void EventListenerJniCallback::OnExternalFileIngested(
DB* db, const ExternalFileIngestionInfo& info) {
if (m_on_external_file_ingested_proxy_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jingestion_info = SetupCallbackInvocation<ExternalFileIngestionInfo>(
env, attached_thread, info,
ExternalFileIngestionInfoJni::fromCppExternalFileIngestionInfo);
if (jingestion_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_external_file_ingested_proxy_mid,
reinterpret_cast<jlong>(db), jingestion_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jingestion_info});
}
void EventListenerJniCallback::OnBackgroundError(BackgroundErrorReason reason,
Status* bg_error) {
if (m_on_background_error_proxy_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jstatus = SetupCallbackInvocation<Status>(
env, attached_thread, *bg_error, StatusJni::construct);
if (jstatus != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_background_error_proxy_mid,
static_cast<jbyte>(reason), jstatus);
}
CleanupCallbackInvocation(env, attached_thread, {&jstatus});
}
void EventListenerJniCallback::OnStallConditionsChanged(
const WriteStallInfo& info) {
if (m_on_stall_conditions_changed_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jwrite_stall_info = SetupCallbackInvocation<WriteStallInfo>(
env, attached_thread, info, WriteStallInfoJni::fromCppWriteStallInfo);
if (jwrite_stall_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_stall_conditions_changed_mid,
jwrite_stall_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jwrite_stall_info});
}
void EventListenerJniCallback::OnFileReadFinish(const FileOperationInfo& info) {
OnFileOperation(m_on_file_read_finish_mid, info);
}
void EventListenerJniCallback::OnFileWriteFinish(
const FileOperationInfo& info) {
OnFileOperation(m_on_file_write_finish_mid, info);
}
void EventListenerJniCallback::OnFileFlushFinish(
const FileOperationInfo& info) {
OnFileOperation(m_on_file_flush_finish_mid, info);
}
void EventListenerJniCallback::OnFileSyncFinish(const FileOperationInfo& info) {
OnFileOperation(m_on_file_sync_finish_mid, info);
}
void EventListenerJniCallback::OnFileRangeSyncFinish(
const FileOperationInfo& info) {
OnFileOperation(m_on_file_range_sync_finish_mid, info);
}
void EventListenerJniCallback::OnFileTruncateFinish(
const FileOperationInfo& info) {
OnFileOperation(m_on_file_truncate_finish_mid, info);
}
void EventListenerJniCallback::OnFileCloseFinish(
const FileOperationInfo& info) {
OnFileOperation(m_on_file_close_finish_mid, info);
}
bool EventListenerJniCallback::ShouldBeNotifiedOnFileIO() {
if (m_should_be_notified_on_file_io == nullptr) {
return false;
}
jboolean attached_thread = JNI_FALSE;
JNIEnv* env = getJniEnv(&attached_thread);
assert(env != nullptr);
jboolean jshould_be_notified =
env->CallBooleanMethod(m_jcallback_obj, m_should_be_notified_on_file_io);
CleanupCallbackInvocation(env, attached_thread, {});
return static_cast<bool>(jshould_be_notified);
}
void EventListenerJniCallback::OnErrorRecoveryBegin(
BackgroundErrorReason reason, Status bg_error, bool* auto_recovery) {
if (m_on_error_recovery_begin_proxy_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jbg_error = SetupCallbackInvocation<Status>(
env, attached_thread, bg_error, StatusJni::construct);
if (jbg_error != nullptr) {
jboolean jauto_recovery = env->CallBooleanMethod(
m_jcallback_obj, m_on_error_recovery_begin_proxy_mid,
static_cast<jbyte>(reason), jbg_error);
*auto_recovery = jauto_recovery == JNI_TRUE;
}
CleanupCallbackInvocation(env, attached_thread, {&jbg_error});
}
void EventListenerJniCallback::OnErrorRecoveryCompleted(Status old_bg_error) {
if (m_on_error_recovery_completed_mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jold_bg_error = SetupCallbackInvocation<Status>(
env, attached_thread, old_bg_error, StatusJni::construct);
if (jold_bg_error != nullptr) {
env->CallVoidMethod(m_jcallback_obj, m_on_error_recovery_completed_mid,
jold_bg_error);
}
CleanupCallbackInvocation(env, attached_thread, {&jold_bg_error});
}
void EventListenerJniCallback::InitCallbackMethodId(
jmethodID& mid, EnabledEventCallback eec, JNIEnv* env,
jmethodID (*get_id)(JNIEnv* env)) {
if (m_enabled_event_callbacks.count(eec) == 1) {
mid = get_id(env);
} else {
mid = nullptr;
}
}
template <class T>
jobject EventListenerJniCallback::SetupCallbackInvocation(
JNIEnv*& env, jboolean& attached_thread, const T& cpp_obj,
jobject (*convert)(JNIEnv* env, const T* cpp_obj)) {
attached_thread = JNI_FALSE;
env = getJniEnv(&attached_thread);
assert(env != nullptr);
return convert(env, &cpp_obj);
}
void EventListenerJniCallback::CleanupCallbackInvocation(
JNIEnv* env, jboolean attached_thread,
std::initializer_list<jobject*> refs) {
for (auto* ref : refs) {
if (*ref == nullptr) continue;
env->DeleteLocalRef(*ref);
}
if (env->ExceptionCheck()) {
// exception thrown from CallVoidMethod
env->ExceptionDescribe(); // print out exception to stderr
}
releaseJniEnv(attached_thread);
}
void EventListenerJniCallback::OnFileOperation(const jmethodID& mid,
const FileOperationInfo& info) {
if (mid == nullptr) {
return;
}
JNIEnv* env;
jboolean attached_thread;
jobject jop_info = SetupCallbackInvocation<FileOperationInfo>(
env, attached_thread, info,
FileOperationInfoJni::fromCppFileOperationInfo);
if (jop_info != nullptr) {
env->CallVoidMethod(m_jcallback_obj, mid, jop_info);
}
CleanupCallbackInvocation(env, attached_thread, {&jop_info});
}
} // namespace rocksdb

122
java/rocksjni/event_listener_jnicallback.h Normal file
View File

@ -0,0 +1,122 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
//
// This file implements the callback "bridge" between Java and C++ for
// rocksdb::EventListener.
#ifndef JAVA_ROCKSJNI_EVENT_LISTENER_JNICALLBACK_H_
#define JAVA_ROCKSJNI_EVENT_LISTENER_JNICALLBACK_H_
#include <jni.h>
#include <memory>
#include <set>
#include "rocksdb/listener.h"
#include "rocksjni/jnicallback.h"
namespace rocksdb {
enum EnabledEventCallback {
ON_FLUSH_COMPLETED = 0x0,
ON_FLUSH_BEGIN = 0x1,
ON_TABLE_FILE_DELETED = 0x2,
ON_COMPACTION_BEGIN = 0x3,
ON_COMPACTION_COMPLETED = 0x4,
ON_TABLE_FILE_CREATED = 0x5,
ON_TABLE_FILE_CREATION_STARTED = 0x6,
ON_MEMTABLE_SEALED = 0x7,
ON_COLUMN_FAMILY_HANDLE_DELETION_STARTED = 0x8,
ON_EXTERNAL_FILE_INGESTED = 0x9,
ON_BACKGROUND_ERROR = 0xA,
ON_STALL_CONDITIONS_CHANGED = 0xB,
ON_FILE_READ_FINISH = 0xC,
ON_FILE_WRITE_FINISH = 0xD,
ON_FILE_FLUSH_FINISH = 0xE,
ON_FILE_SYNC_FINISH = 0xF,
ON_FILE_RANGE_SYNC_FINISH = 0x10,
ON_FILE_TRUNCATE_FINISH = 0x11,
ON_FILE_CLOSE_FINISH = 0x12,
SHOULD_BE_NOTIFIED_ON_FILE_IO = 0x13,
ON_ERROR_RECOVERY_BEGIN = 0x14,
ON_ERROR_RECOVERY_COMPLETED = 0x15,
NUM_ENABLED_EVENT_CALLBACK = 0x16,
};
class EventListenerJniCallback : public JniCallback, public EventListener {
public:
EventListenerJniCallback(
JNIEnv* env, jobject jevent_listener,
const std::set<EnabledEventCallback>& enabled_event_callbacks);
virtual ~EventListenerJniCallback();
virtual void OnFlushCompleted(DB* db, const FlushJobInfo& flush_job_info);
virtual void OnFlushBegin(DB* db, const FlushJobInfo& flush_job_info);
virtual void OnTableFileDeleted(const TableFileDeletionInfo& info);
virtual void OnCompactionBegin(DB* db, const CompactionJobInfo& ci);
virtual void OnCompactionCompleted(DB* db, const CompactionJobInfo& ci);
virtual void OnTableFileCreated(const TableFileCreationInfo& info);
virtual void OnTableFileCreationStarted(
const TableFileCreationBriefInfo& info);
virtual void OnMemTableSealed(const MemTableInfo& info);
virtual void OnColumnFamilyHandleDeletionStarted(ColumnFamilyHandle* handle);
virtual void OnExternalFileIngested(DB* db,
const ExternalFileIngestionInfo& info);
virtual void OnBackgroundError(BackgroundErrorReason reason,
Status* bg_error);
virtual void OnStallConditionsChanged(const WriteStallInfo& info);
virtual void OnFileReadFinish(const FileOperationInfo& info);
virtual void OnFileWriteFinish(const FileOperationInfo& info);
virtual void OnFileFlushFinish(const FileOperationInfo& info);
virtual void OnFileSyncFinish(const FileOperationInfo& info);
virtual void OnFileRangeSyncFinish(const FileOperationInfo& info);
virtual void OnFileTruncateFinish(const FileOperationInfo& info);
virtual void OnFileCloseFinish(const FileOperationInfo& info);
virtual bool ShouldBeNotifiedOnFileIO();
virtual void OnErrorRecoveryBegin(BackgroundErrorReason reason,
Status bg_error, bool* auto_recovery);
virtual void OnErrorRecoveryCompleted(Status old_bg_error);
private:
inline void InitCallbackMethodId(jmethodID& mid, EnabledEventCallback eec,
JNIEnv* env,
jmethodID (*get_id)(JNIEnv* env));
template <class T>
inline jobject SetupCallbackInvocation(
JNIEnv*& env, jboolean& attached_thread, const T& cpp_obj,
jobject (*convert)(JNIEnv* env, const T* cpp_obj));
inline void CleanupCallbackInvocation(JNIEnv* env, jboolean attached_thread,
std::initializer_list<jobject*> refs);
inline void OnFileOperation(const jmethodID& mid,
const FileOperationInfo& info);
const std::set<EnabledEventCallback> m_enabled_event_callbacks;
jmethodID m_on_flush_completed_proxy_mid;
jmethodID m_on_flush_begin_proxy_mid;
jmethodID m_on_table_file_deleted_mid;
jmethodID m_on_compaction_begin_proxy_mid;
jmethodID m_on_compaction_completed_proxy_mid;
jmethodID m_on_table_file_created_mid;
jmethodID m_on_table_file_creation_started_mid;
jmethodID m_on_mem_table_sealed_mid;
jmethodID m_on_column_family_handle_deletion_started_mid;
jmethodID m_on_external_file_ingested_proxy_mid;
jmethodID m_on_background_error_proxy_mid;
jmethodID m_on_stall_conditions_changed_mid;
jmethodID m_on_file_read_finish_mid;
jmethodID m_on_file_write_finish_mid;
jmethodID m_on_file_flush_finish_mid;
jmethodID m_on_file_sync_finish_mid;
jmethodID m_on_file_range_sync_finish_mid;
jmethodID m_on_file_truncate_finish_mid;
jmethodID m_on_file_close_finish_mid;
jmethodID m_should_be_notified_on_file_io;
jmethodID m_on_error_recovery_begin_proxy_mid;
jmethodID m_on_error_recovery_completed_mid;
};
} // namespace rocksdb
#endif // JAVA_ROCKSJNI_EVENT_LISTENER_JNICALLBACK_H_

2
java/rocksjni/jnicallback.h
View File

@ -19,6 +19,8 @@ class JniCallback {
JniCallback(JNIEnv* env, jobject jcallback_obj);
virtual ~JniCallback();
const jobject& GetJavaObject() const { return m_jcallback_obj; }
protected:
JavaVM* m_jvm;
jobject m_jcallback_obj;

93
java/rocksjni/options.cc
View File

@ -1767,6 +1767,76 @@ jboolean Java_org_rocksdb_Options_strictBytesPerSync(
return static_cast<jboolean>(opt->strict_bytes_per_sync);
}
// Note: the RocksJava API currently only supports EventListeners implemented in
// Java. It could be extended in future to also support adding/removing
// EventListeners implemented in C++.
static void rocksdb_set_event_listeners_helper(
JNIEnv* env, jlongArray jlistener_array,
std::vector<std::shared_ptr<ROCKSDB_NAMESPACE::EventListener>>&
listener_sptr_vec) {
jlong* ptr_jlistener_array =
env->GetLongArrayElements(jlistener_array, nullptr);
if (ptr_jlistener_array == nullptr) {
// exception thrown: OutOfMemoryError
return;
}
const jsize array_size = env->GetArrayLength(jlistener_array);
listener_sptr_vec.clear();
for (jsize i = 0; i < array_size; ++i) {
const auto& listener_sptr =
*reinterpret_cast<std::shared_ptr<ROCKSDB_NAMESPACE::EventListener>*>(
ptr_jlistener_array[i]);
listener_sptr_vec.push_back(listener_sptr);
}
}
/*
* Class: org_rocksdb_Options
* Method: setEventListeners
* Signature: (J[J)V
*/
void Java_org_rocksdb_Options_setEventListeners(JNIEnv* env, jclass,
jlong jhandle,
jlongArray jlistener_array) {
auto* opt = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle);
rocksdb_set_event_listeners_helper(env, jlistener_array, opt->listeners);
}
// Note: the RocksJava API currently only supports EventListeners implemented in
// Java. It could be extended in future to also support adding/removing
// EventListeners implemented in C++.
static jobjectArray rocksdb_get_event_listeners_helper(
JNIEnv* env,
const std::vector<std::shared_ptr<ROCKSDB_NAMESPACE::EventListener>>&
listener_sptr_vec) {
jsize sz = static_cast<jsize>(listener_sptr_vec.size());
jclass jlistener_clazz =
ROCKSDB_NAMESPACE::AbstractEventListenerJni::getJClass(env);
jobjectArray jlisteners = env->NewObjectArray(sz, jlistener_clazz, nullptr);
if (jlisteners == nullptr) {
// exception thrown: OutOfMemoryError
return nullptr;
}
for (jsize i = 0; i < sz; ++i) {
const auto* jni_cb =
static_cast<ROCKSDB_NAMESPACE::EventListenerJniCallback*>(
listener_sptr_vec[i].get());
env->SetObjectArrayElement(jlisteners, i, jni_cb->GetJavaObject());
}
return jlisteners;
}
/*
* Class: org_rocksdb_Options
* Method: eventListeners
* Signature: (J)[Lorg/rocksdb/AbstractEventListener;
*/
jobjectArray Java_org_rocksdb_Options_eventListeners(JNIEnv* env, jclass,
jlong jhandle) {
auto* opt = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle);
return rocksdb_get_event_listeners_helper(env, opt->listeners);
}
/*
* Class: org_rocksdb_Options
* Method: setEnableThreadTracking
@ -6549,6 +6619,29 @@ jboolean Java_org_rocksdb_DBOptions_strictBytesPerSync(
->strict_bytes_per_sync);
}
/*
* Class: org_rocksdb_DBOptions
* Method: setEventListeners
* Signature: (J[J)V
*/
void Java_org_rocksdb_DBOptions_setEventListeners(JNIEnv* env, jclass,
jlong jhandle,
jlongArray jlistener_array) {
auto* opt = reinterpret_cast<ROCKSDB_NAMESPACE::DBOptions*>(jhandle);
rocksdb_set_event_listeners_helper(env, jlistener_array, opt->listeners);
}
/*
* Class: org_rocksdb_DBOptions
* Method: eventListeners
* Signature: (J)[Lorg/rocksdb/AbstractEventListener;
*/
jobjectArray Java_org_rocksdb_DBOptions_eventListeners(JNIEnv* env, jclass,
jlong jhandle) {
auto* opt = reinterpret_cast<ROCKSDB_NAMESPACE::DBOptions*>(jhandle);
return rocksdb_get_event_listeners_helper(env, opt->listeners);
}
/*
* Class: org_rocksdb_DBOptions
* Method: setDelayedWriteRate

777
java/rocksjni/portal.h
View File

@ -10,14 +10,16 @@
#ifndef JAVA_ROCKSJNI_PORTAL_H_
#define JAVA_ROCKSJNI_PORTAL_H_
#include <jni.h>
#include <algorithm>
#include <cstring>
#include <functional>
#include <iostream>
#include <iterator>
#include <jni.h>
#include <limits>
#include <memory>
#include <set>
#include <string>
#include <type_traits>
#include <vector>
@ -34,6 +36,7 @@
#include "rocksdb/utilities/write_batch_with_index.h"
#include "rocksjni/compaction_filter_factory_jnicallback.h"
#include "rocksjni/comparatorjnicallback.h"
#include "rocksjni/event_listener_jnicallback.h"
#include "rocksjni/loggerjnicallback.h"
#include "rocksjni/table_filter_jnicallback.h"
#include "rocksjni/trace_writer_jnicallback.h"
@ -438,6 +441,10 @@ class StatusJni
return jstatus;
}
static jobject construct(JNIEnv* env, const Status* status) {
return construct(env, *status);
}
// Returns the equivalent org.rocksdb.Status.Code for the provided
// C++ ROCKSDB_NAMESPACE::Status::Code enum
static jbyte toJavaStatusCode(const ROCKSDB_NAMESPACE::Status::Code& code) {
@ -3461,6 +3468,19 @@ class ColumnFamilyHandleJni
: public RocksDBNativeClass<ROCKSDB_NAMESPACE::ColumnFamilyHandle*,
ColumnFamilyHandleJni> {
public:
static jobject fromCppColumnFamilyHandle(
JNIEnv* env, const ROCKSDB_NAMESPACE::ColumnFamilyHandle* info) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
return env->NewObject(jclazz, ctor, reinterpret_cast<jlong>(info));
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(clazz, "<init>", "(J)V");
}
/**
* Get the Java Class org.rocksdb.ColumnFamilyHandle
*
@ -7659,5 +7679,760 @@ class SanityLevelJni {
}
}
};
// The portal class for org.rocksdb.AbstractListener.EnabledEventCallback
class EnabledEventCallbackJni {
public:
// Returns the set of equivalent C++
// rocksdb::EnabledEventCallbackJni::EnabledEventCallback enums for
// the provided Java jenabled_event_callback_values
static std::set<EnabledEventCallback> toCppEnabledEventCallbacks(
jlong jenabled_event_callback_values) {
std::set<EnabledEventCallback> enabled_event_callbacks;
for (size_t i = 0; i < EnabledEventCallback::NUM_ENABLED_EVENT_CALLBACK;
++i) {
if (((1ULL << i) & jenabled_event_callback_values) > 0) {
enabled_event_callbacks.emplace(static_cast<EnabledEventCallback>(i));
}
}
return enabled_event_callbacks;
}
};
// The portal class for org.rocksdb.AbstractEventListener
class AbstractEventListenerJni
: public RocksDBNativeClass<
const ROCKSDB_NAMESPACE::EventListenerJniCallback*,
AbstractEventListenerJni> {
public:
/**
* Get the Java Class org.rocksdb.AbstractEventListener
*
* @param env A pointer to the Java environment
*
* @return The Java Class or nullptr if one of the
* ClassFormatError, ClassCircularityError, NoClassDefFoundError,
* OutOfMemoryError or ExceptionInInitializerError exceptions is thrown
*/
static jclass getJClass(JNIEnv* env) {
return RocksDBNativeClass::getJClass(env,
"org/rocksdb/AbstractEventListener");
}
/**
* Get the Java Method: AbstractEventListener#onFlushCompletedProxy
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFlushCompletedProxyMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(jclazz, "onFlushCompletedProxy",
"(JLorg/rocksdb/FlushJobInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onFlushBeginProxy
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFlushBeginProxyMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(jclazz, "onFlushBeginProxy",
"(JLorg/rocksdb/FlushJobInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onTableFileDeleted
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnTableFileDeletedMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "onTableFileDeleted", "(Lorg/rocksdb/TableFileDeletionInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onCompactionBeginProxy
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnCompactionBeginProxyMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid =
env->GetMethodID(jclazz, "onCompactionBeginProxy",
"(JLorg/rocksdb/CompactionJobInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onCompactionCompletedProxy
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnCompactionCompletedProxyMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid =
env->GetMethodID(jclazz, "onCompactionCompletedProxy",
"(JLorg/rocksdb/CompactionJobInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onTableFileCreated
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnTableFileCreatedMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "onTableFileCreated", "(Lorg/rocksdb/TableFileCreationInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onTableFileCreationStarted
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnTableFileCreationStartedMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid =
env->GetMethodID(jclazz, "onTableFileCreationStarted",
"(Lorg/rocksdb/TableFileCreationBriefInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onMemTableSealed
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnMemTableSealedMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(jclazz, "onMemTableSealed",
"(Lorg/rocksdb/MemTableInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method:
* AbstractEventListener#onColumnFamilyHandleDeletionStarted
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnColumnFamilyHandleDeletionStartedMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid =
env->GetMethodID(jclazz, "onColumnFamilyHandleDeletionStarted",
"(Lorg/rocksdb/ColumnFamilyHandle;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onExternalFileIngestedProxy
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnExternalFileIngestedProxyMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid =
env->GetMethodID(jclazz, "onExternalFileIngestedProxy",
"(JLorg/rocksdb/ExternalFileIngestionInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onBackgroundError
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnBackgroundErrorProxyMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(jclazz, "onBackgroundErrorProxy",
"(BLorg/rocksdb/Status;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onStallConditionsChanged
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnStallConditionsChangedMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(jclazz, "onStallConditionsChanged",
"(Lorg/rocksdb/WriteStallInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onFileReadFinish
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFileReadFinishMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "onFileReadFinish", "(Lorg/rocksdb/FileOperationInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onFileWriteFinish
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFileWriteFinishMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "onFileWriteFinish", "(Lorg/rocksdb/FileOperationInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#OnFileFlushFinish
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFileFlushFinishMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "OnFileFlushFinish", "(Lorg/rocksdb/FileOperationInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#OnFileSyncFinish
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFileSyncFinishMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "OnFileSyncFinish", "(Lorg/rocksdb/FileOperationInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#OnFileRangeSyncFinish
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFileRangeSyncFinishMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "OnFileRangeSyncFinish", "(Lorg/rocksdb/FileOperationInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#OnFileTruncateFinish
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFileTruncateFinishMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "OnFileTruncateFinish", "(Lorg/rocksdb/FileOperationInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#OnFileCloseFinish
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnFileCloseFinishMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(
jclazz, "OnFileCloseFinish", "(Lorg/rocksdb/FileOperationInfo;)V");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#shouldBeNotifiedOnFileIO
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getShouldBeNotifiedOnFileIOMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid =
env->GetMethodID(jclazz, "shouldBeNotifiedOnFileIO", "()Z");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onErrorRecoveryBeginProxy
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnErrorRecoveryBeginProxyMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(jclazz, "onErrorRecoveryBeginProxy",
"(BLorg/rocksdb/Status;)Z");
assert(mid != nullptr);
return mid;
}
/**
* Get the Java Method: AbstractEventListener#onErrorRecoveryCompleted
*
* @param env A pointer to the Java environment
*
* @return The Java Method ID
*/
static jmethodID getOnErrorRecoveryCompletedMethodId(JNIEnv* env) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID mid = env->GetMethodID(jclazz, "onErrorRecoveryCompleted",
"(Lorg/rocksdb/Status;)V");
assert(mid != nullptr);
return mid;
}
};
class FlushJobInfoJni : public JavaClass {
public:
/**
* Create a new Java org.rocksdb.FlushJobInfo object.
*
* @param env A pointer to the Java environment
* @param flush_job_info A Cpp flush job info object
*
* @return A reference to a Java org.rocksdb.FlushJobInfo object, or
* nullptr if an an exception occurs
*/
static jobject fromCppFlushJobInfo(
JNIEnv* env, const ROCKSDB_NAMESPACE::FlushJobInfo* flush_job_info) {
jclass jclazz = getJClass(env);
if (jclazz == nullptr) {
// exception occurred accessing class
return nullptr;
}
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
jstring jcf_name = JniUtil::toJavaString(env, &flush_job_info->cf_name);
if (env->ExceptionCheck()) {
return nullptr;
}
jstring jfile_path = JniUtil::toJavaString(env, &flush_job_info->file_path);
if (env->ExceptionCheck()) {
env->DeleteLocalRef(jfile_path);
return nullptr;
}
jobject jtable_properties = TablePropertiesJni::fromCppTableProperties(
env, flush_job_info->table_properties);
if (jtable_properties == nullptr) {
env->DeleteLocalRef(jcf_name);
env->DeleteLocalRef(jfile_path);
return nullptr;
}
return env->NewObject(
jclazz, ctor, static_cast<jlong>(flush_job_info->cf_id), jcf_name,
jfile_path, static_cast<jlong>(flush_job_info->thread_id),
static_cast<jint>(flush_job_info->job_id),
static_cast<jboolean>(flush_job_info->triggered_writes_slowdown),
static_cast<jboolean>(flush_job_info->triggered_writes_stop),
static_cast<jlong>(flush_job_info->smallest_seqno),
static_cast<jlong>(flush_job_info->largest_seqno), jtable_properties,
static_cast<jbyte>(flush_job_info->flush_reason));
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/FlushJobInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(clazz, "<init>",
"(JLjava/lang/String;Ljava/lang/String;JIZZJJLorg/"
"rocksdb/TableProperties;B)V");
}
};
class TableFileDeletionInfoJni : public JavaClass {
public:
/**
* Create a new Java org.rocksdb.TableFileDeletionInfo object.
*
* @param env A pointer to the Java environment
* @param file_del_info A Cpp table file deletion info object
*
* @return A reference to a Java org.rocksdb.TableFileDeletionInfo object, or
* nullptr if an an exception occurs
*/
static jobject fromCppTableFileDeletionInfo(
JNIEnv* env,
const ROCKSDB_NAMESPACE::TableFileDeletionInfo* file_del_info) {
jclass jclazz = getJClass(env);
if (jclazz == nullptr) {
// exception occurred accessing class
return nullptr;
}
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
jstring jdb_name = JniUtil::toJavaString(env, &file_del_info->db_name);
if (env->ExceptionCheck()) {
return nullptr;
}
jobject jstatus = StatusJni::construct(env, file_del_info->status);
if (jstatus == nullptr) {
env->DeleteLocalRef(jdb_name);
return nullptr;
}
return env->NewObject(jclazz, ctor, jdb_name,
JniUtil::toJavaString(env, &file_del_info->file_path),
static_cast<jint>(file_del_info->job_id), jstatus);
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/TableFileDeletionInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(
clazz, "<init>",
"(Ljava/lang/String;Ljava/lang/String;ILorg/rocksdb/Status;)V");
}
};
class CompactionJobInfoJni : public JavaClass {
public:
static jobject fromCppCompactionJobInfo(
JNIEnv* env,
const ROCKSDB_NAMESPACE::CompactionJobInfo* compaction_job_info) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
return env->NewObject(jclazz, ctor,
reinterpret_cast<jlong>(compaction_job_info));
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/CompactionJobInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(clazz, "<init>", "(J)V");
}
};
class TableFileCreationInfoJni : public JavaClass {
public:
static jobject fromCppTableFileCreationInfo(
JNIEnv* env, const ROCKSDB_NAMESPACE::TableFileCreationInfo* info) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
jstring jdb_name = JniUtil::toJavaString(env, &info->db_name);
if (env->ExceptionCheck()) {
return nullptr;
}
jstring jcf_name = JniUtil::toJavaString(env, &info->cf_name);
if (env->ExceptionCheck()) {
env->DeleteLocalRef(jdb_name);
return nullptr;
}
jstring jfile_path = JniUtil::toJavaString(env, &info->file_path);
if (env->ExceptionCheck()) {
env->DeleteLocalRef(jdb_name);
env->DeleteLocalRef(jcf_name);
return nullptr;
}
jobject jtable_properties =
TablePropertiesJni::fromCppTableProperties(env, info->table_properties);
if (jtable_properties == nullptr) {
env->DeleteLocalRef(jdb_name);
env->DeleteLocalRef(jcf_name);
return nullptr;
}
jobject jstatus = StatusJni::construct(env, info->status);
if (jstatus == nullptr) {
env->DeleteLocalRef(jdb_name);
env->DeleteLocalRef(jcf_name);
env->DeleteLocalRef(jtable_properties);
return nullptr;
}
return env->NewObject(jclazz, ctor, static_cast<jlong>(info->file_size),
jtable_properties, jstatus, jdb_name, jcf_name,
jfile_path, static_cast<jint>(info->job_id),
static_cast<jbyte>(info->reason));
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/TableFileCreationInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(
clazz, "<init>",
"(JLorg/rocksdb/TableProperties;Lorg/rocksdb/Status;Ljava/lang/"
"String;Ljava/lang/String;Ljava/lang/String;IB)V");
}
};
class TableFileCreationBriefInfoJni : public JavaClass {
public:
static jobject fromCppTableFileCreationBriefInfo(
JNIEnv* env, const ROCKSDB_NAMESPACE::TableFileCreationBriefInfo* info) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
jstring jdb_name = JniUtil::toJavaString(env, &info->db_name);
if (env->ExceptionCheck()) {
return nullptr;
}
jstring jcf_name = JniUtil::toJavaString(env, &info->cf_name);
if (env->ExceptionCheck()) {
env->DeleteLocalRef(jdb_name);
return nullptr;
}
jstring jfile_path = JniUtil::toJavaString(env, &info->file_path);
if (env->ExceptionCheck()) {
env->DeleteLocalRef(jdb_name);
env->DeleteLocalRef(jcf_name);
return nullptr;
}
return env->NewObject(jclazz, ctor, jdb_name, jcf_name, jfile_path,
static_cast<jint>(info->job_id),
static_cast<jbyte>(info->reason));
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/TableFileCreationBriefInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(
clazz, "<init>",
"(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;IB)V");
}
};
class MemTableInfoJni : public JavaClass {
public:
static jobject fromCppMemTableInfo(
JNIEnv* env, const ROCKSDB_NAMESPACE::MemTableInfo* info) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
jstring jcf_name = JniUtil::toJavaString(env, &info->cf_name);
if (env->ExceptionCheck()) {
return nullptr;
}
return env->NewObject(jclazz, ctor, jcf_name,
static_cast<jlong>(info->first_seqno),
static_cast<jlong>(info->earliest_seqno),
static_cast<jlong>(info->num_entries),
static_cast<jlong>(info->num_deletes));
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/MemTableInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(clazz, "<init>", "(Ljava/lang/String;JJJJ)V");
}
};
class ExternalFileIngestionInfoJni : public JavaClass {
public:
static jobject fromCppExternalFileIngestionInfo(
JNIEnv* env, const ROCKSDB_NAMESPACE::ExternalFileIngestionInfo* info) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
jstring jcf_name = JniUtil::toJavaString(env, &info->cf_name);
if (env->ExceptionCheck()) {
return nullptr;
}
jstring jexternal_file_path =
JniUtil::toJavaString(env, &info->external_file_path);
if (env->ExceptionCheck()) {
env->DeleteLocalRef(jcf_name);
return nullptr;
}
jstring jinternal_file_path =
JniUtil::toJavaString(env, &info->internal_file_path);
if (env->ExceptionCheck()) {
env->DeleteLocalRef(jcf_name);
env->DeleteLocalRef(jexternal_file_path);
return nullptr;
}
jobject jtable_properties =
TablePropertiesJni::fromCppTableProperties(env, info->table_properties);
if (jtable_properties == nullptr) {
env->DeleteLocalRef(jcf_name);
env->DeleteLocalRef(jexternal_file_path);
env->DeleteLocalRef(jinternal_file_path);
return nullptr;
}
return env->NewObject(
jclazz, ctor, jcf_name, jexternal_file_path, jinternal_file_path,
static_cast<jlong>(info->global_seqno), jtable_properties);
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/ExternalFileIngestionInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(clazz, "<init>",
"(Ljava/lang/String;Ljava/lang/String;Ljava/lang/"
"String;JLorg/rocksdb/TableProperties;)V");
}
};
class WriteStallInfoJni : public JavaClass {
public:
static jobject fromCppWriteStallInfo(
JNIEnv* env, const ROCKSDB_NAMESPACE::WriteStallInfo* info) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
jstring jcf_name = JniUtil::toJavaString(env, &info->cf_name);
if (env->ExceptionCheck()) {
return nullptr;
}
return env->NewObject(jclazz, ctor, jcf_name,
static_cast<jbyte>(info->condition.cur),
static_cast<jbyte>(info->condition.prev));
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/WriteStallInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(clazz, "<init>", "(Ljava/lang/String;BB)V");
}
};
class FileOperationInfoJni : public JavaClass {
public:
static jobject fromCppFileOperationInfo(
JNIEnv* env, const ROCKSDB_NAMESPACE::FileOperationInfo* info) {
jclass jclazz = getJClass(env);
assert(jclazz != nullptr);
static jmethodID ctor = getConstructorMethodId(env, jclazz);
assert(ctor != nullptr);
jstring jpath = JniUtil::toJavaString(env, &info->path);
if (env->ExceptionCheck()) {
return nullptr;
}
jobject jstatus = StatusJni::construct(env, info->status);
if (jstatus == nullptr) {
env->DeleteLocalRef(jpath);
return nullptr;
}
return env->NewObject(
jclazz, ctor, jpath, static_cast<jlong>(info->offset),
static_cast<jlong>(info->length),
static_cast<jlong>(info->start_ts.time_since_epoch().count()),
static_cast<jlong>(info->duration.count()), jstatus);
}
static jclass getJClass(JNIEnv* env) {
return JavaClass::getJClass(env, "org/rocksdb/FileOperationInfo");
}
static jmethodID getConstructorMethodId(JNIEnv* env, jclass clazz) {
return env->GetMethodID(clazz, "<init>",
"(Ljava/lang/String;JJJJLorg/rocksdb/Status;)V");
}
};
} // namespace ROCKSDB_NAMESPACE
#endif // JAVA_ROCKSJNI_PORTAL_H_

189
java/rocksjni/testable_event_listener.cc Normal file
View File

@ -0,0 +1,189 @@
#include <climits>
#include <utility>
#include "include/org_rocksdb_test_TestableEventListener.h"
#include "rocksdb/listener.h"
using namespace ROCKSDB_NAMESPACE;
static TableProperties newTablePropertiesForTest() {
TableProperties table_properties;
table_properties.data_size = LLONG_MAX;
table_properties.index_size = LLONG_MAX;
table_properties.index_partitions = LLONG_MAX;
table_properties.top_level_index_size = LLONG_MAX;
table_properties.index_key_is_user_key = LLONG_MAX;
table_properties.index_value_is_delta_encoded = LLONG_MAX;
table_properties.filter_size = LLONG_MAX;
table_properties.raw_key_size = LLONG_MAX;
table_properties.raw_value_size = LLONG_MAX;
table_properties.num_data_blocks = LLONG_MAX;
table_properties.num_entries = LLONG_MAX;
table_properties.num_deletions = LLONG_MAX;
table_properties.num_merge_operands = LLONG_MAX;
table_properties.num_range_deletions = LLONG_MAX;
table_properties.format_version = LLONG_MAX;
table_properties.fixed_key_len = LLONG_MAX;
table_properties.column_family_id = LLONG_MAX;
table_properties.creation_time = LLONG_MAX;
table_properties.oldest_key_time = LLONG_MAX;
table_properties.file_creation_time = LLONG_MAX;
table_properties.db_id = "dbId";
table_properties.db_session_id = "sessionId";
table_properties.column_family_name = "columnFamilyName";
table_properties.filter_policy_name = "filterPolicyName";
table_properties.comparator_name = "comparatorName";
table_properties.merge_operator_name = "mergeOperatorName";
table_properties.prefix_extractor_name = "prefixExtractorName";
table_properties.property_collectors_names = "propertyCollectorsNames";
table_properties.compression_name = "compressionName";
table_properties.compression_options = "compressionOptions";
table_properties.user_collected_properties = {{"key", "value"}};
table_properties.readable_properties = {{"key", "value"}};
table_properties.properties_offsets = {{"key", LLONG_MAX}};
return table_properties;
}
/*
* Class: org_rocksdb_test_TestableEventListener
* Method: invokeAllCallbacks
* Signature: (J)V
*/
void Java_org_rocksdb_test_TestableEventListener_invokeAllCallbacks(
JNIEnv *, jclass, jlong jhandle) {
const auto &el =
*reinterpret_cast<std::shared_ptr<ROCKSDB_NAMESPACE::EventListener> *>(
jhandle);
TableProperties table_properties = newTablePropertiesForTest();
FlushJobInfo flush_job_info;
flush_job_info.cf_id = INT_MAX;
flush_job_info.cf_name = "testColumnFamily";
flush_job_info.file_path = "/file/path";
flush_job_info.file_number = LLONG_MAX;
flush_job_info.oldest_blob_file_number = LLONG_MAX;
flush_job_info.thread_id = LLONG_MAX;
flush_job_info.job_id = INT_MAX;
flush_job_info.triggered_writes_slowdown = true;
flush_job_info.triggered_writes_stop = true;
flush_job_info.smallest_seqno = LLONG_MAX;
flush_job_info.largest_seqno = LLONG_MAX;
flush_job_info.table_properties = table_properties;
flush_job_info.flush_reason = FlushReason::kManualFlush;
el->OnFlushCompleted(nullptr, flush_job_info);
el->OnFlushBegin(nullptr, flush_job_info);
Status status = Status::Incomplete(Status::SubCode::kNoSpace);
TableFileDeletionInfo file_deletion_info;
file_deletion_info.db_name = "dbName";
file_deletion_info.file_path = "/file/path";
file_deletion_info.job_id = INT_MAX;
file_deletion_info.status = status;
el->OnTableFileDeleted(file_deletion_info);
CompactionJobInfo compaction_job_info;
compaction_job_info.cf_id = INT_MAX;
compaction_job_info.cf_name = "compactionColumnFamily";
compaction_job_info.status = status;
compaction_job_info.thread_id = LLONG_MAX;
compaction_job_info.job_id = INT_MAX;
compaction_job_info.base_input_level = INT_MAX;
compaction_job_info.output_level = INT_MAX;
compaction_job_info.input_files = {"inputFile.sst"};
compaction_job_info.input_file_infos = {};
compaction_job_info.output_files = {"outputFile.sst"};
compaction_job_info.output_file_infos = {};
compaction_job_info.table_properties = {
{"tableProperties", std::shared_ptr<TableProperties>(
&table_properties, [](TableProperties *) {})}};
compaction_job_info.compaction_reason = CompactionReason::kFlush;
compaction_job_info.compression = CompressionType::kSnappyCompression;
compaction_job_info.stats = CompactionJobStats();
el->OnCompactionBegin(nullptr, compaction_job_info);
el->OnCompactionCompleted(nullptr, compaction_job_info);
TableFileCreationInfo file_creation_info;
file_creation_info.file_size = LLONG_MAX;
file_creation_info.table_properties = table_properties;
file_creation_info.status = status;
file_creation_info.file_checksum = "fileChecksum";
file_creation_info.file_checksum_func_name = "fileChecksumFuncName";
file_creation_info.db_name = "dbName";
file_creation_info.cf_name = "columnFamilyName";
file_creation_info.file_path = "/file/path";
file_creation_info.job_id = INT_MAX;
file_creation_info.reason = TableFileCreationReason::kMisc;
el->OnTableFileCreated(file_creation_info);
TableFileCreationBriefInfo file_creation_brief_info;
file_creation_brief_info.db_name = "dbName";
file_creation_brief_info.cf_name = "columnFamilyName";
file_creation_brief_info.file_path = "/file/path";
file_creation_brief_info.job_id = INT_MAX;
file_creation_brief_info.reason = TableFileCreationReason::kMisc;
el->OnTableFileCreationStarted(file_creation_brief_info);
MemTableInfo mem_table_info;
mem_table_info.cf_name = "columnFamilyName";
mem_table_info.first_seqno = LLONG_MAX;
mem_table_info.earliest_seqno = LLONG_MAX;
mem_table_info.num_entries = LLONG_MAX;
mem_table_info.num_deletes = LLONG_MAX;
el->OnMemTableSealed(mem_table_info);
el->OnColumnFamilyHandleDeletionStarted(nullptr);
ExternalFileIngestionInfo file_ingestion_info;
file_ingestion_info.cf_name = "columnFamilyName";
file_ingestion_info.external_file_path = "/external/file/path";
file_ingestion_info.internal_file_path = "/internal/file/path";
file_ingestion_info.global_seqno = LLONG_MAX;
file_ingestion_info.table_properties = table_properties;
el->OnExternalFileIngested(nullptr, file_ingestion_info);
el->OnBackgroundError(BackgroundErrorReason::kFlush, &status);
WriteStallInfo write_stall_info;
write_stall_info.cf_name = "columnFamilyName";
write_stall_info.condition.cur = WriteStallCondition::kDelayed;
write_stall_info.condition.prev = WriteStallCondition::kStopped;
el->OnStallConditionsChanged(write_stall_info);
FileOperationInfo op_info = FileOperationInfo(
FileOperationType::kRead, "/file/path",
std::make_pair(std::chrono::time_point<std::chrono::system_clock,
std::chrono::nanoseconds>(
std::chrono::nanoseconds(1600699420000000000ll)),
std::chrono::time_point<std::chrono::steady_clock,
std::chrono::nanoseconds>(
std::chrono::nanoseconds(1600699420000000000ll))),
std::chrono::time_point<std::chrono::steady_clock,
std::chrono::nanoseconds>(
std::chrono::nanoseconds(1600699425000000000ll)),
status);
op_info.offset = LLONG_MAX;
op_info.length = LLONG_MAX;
op_info.status = status;
el->OnFileReadFinish(op_info);
el->OnFileWriteFinish(op_info);
el->OnFileFlushFinish(op_info);
el->OnFileSyncFinish(op_info);
el->OnFileRangeSyncFinish(op_info);
el->OnFileTruncateFinish(op_info);
el->OnFileCloseFinish(op_info);
el->ShouldBeNotifiedOnFileIO();
bool auto_recovery;
el->OnErrorRecoveryBegin(BackgroundErrorReason::kFlush, status,
&auto_recovery);
el->OnErrorRecoveryCompleted(status);
}

334
java/src/main/java/org/rocksdb/AbstractEventListener.java Normal file
View File

@ -0,0 +1,334 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import static org.rocksdb.AbstractEventListener.EnabledEventCallback.*;
/**
* Base class for Event Listeners.
*/
public abstract class AbstractEventListener extends RocksCallbackObject implements EventListener {
public enum EnabledEventCallback {
ON_FLUSH_COMPLETED((byte) 0x0),
ON_FLUSH_BEGIN((byte) 0x1),
ON_TABLE_FILE_DELETED((byte) 0x2),
ON_COMPACTION_BEGIN((byte) 0x3),
ON_COMPACTION_COMPLETED((byte) 0x4),
ON_TABLE_FILE_CREATED((byte) 0x5),
ON_TABLE_FILE_CREATION_STARTED((byte) 0x6),
ON_MEMTABLE_SEALED((byte) 0x7),
ON_COLUMN_FAMILY_HANDLE_DELETION_STARTED((byte) 0x8),
ON_EXTERNAL_FILE_INGESTED((byte) 0x9),
ON_BACKGROUND_ERROR((byte) 0xA),
ON_STALL_CONDITIONS_CHANGED((byte) 0xB),
ON_FILE_READ_FINISH((byte) 0xC),
ON_FILE_WRITE_FINISH((byte) 0xD),
ON_FILE_FLUSH_FINISH((byte) 0xE),
ON_FILE_SYNC_FINISH((byte) 0xF),
ON_FILE_RANGE_SYNC_FINISH((byte) 0x10),
ON_FILE_TRUNCATE_FINISH((byte) 0x11),
ON_FILE_CLOSE_FINISH((byte) 0x12),
SHOULD_BE_NOTIFIED_ON_FILE_IO((byte) 0x13),
ON_ERROR_RECOVERY_BEGIN((byte) 0x14),
ON_ERROR_RECOVERY_COMPLETED((byte) 0x15);
private final byte value;
EnabledEventCallback(final byte value) {
this.value = value;
}
/**
* Get the internal representation value.
*
* @return the internal representation value
*/
byte getValue() {
return value;
}
/**
* Get the EnabledEventCallbacks from the internal representation value.
*
* @return the enabled event callback.
*
* @throws IllegalArgumentException if the value is unknown.
*/
static EnabledEventCallback fromValue(final byte value) {
for (final EnabledEventCallback enabledEventCallback : EnabledEventCallback.values()) {
if (enabledEventCallback.value == value) {
return enabledEventCallback;
}
}
throw new IllegalArgumentException(
"Illegal value provided for EnabledEventCallback: " + value);
}
}
/**
* Creates an Event Listener that will
* received all callbacks from C++.
*
* If you don't need all callbacks, it is much more efficient to
* just register for the ones you need by calling
* {@link #AbstractEventListener(EnabledEventCallback...)} instead.
*/
protected AbstractEventListener() {
this(ON_FLUSH_COMPLETED, ON_FLUSH_BEGIN, ON_TABLE_FILE_DELETED, ON_COMPACTION_BEGIN,
ON_COMPACTION_COMPLETED, ON_TABLE_FILE_CREATED, ON_TABLE_FILE_CREATION_STARTED,
ON_MEMTABLE_SEALED, ON_COLUMN_FAMILY_HANDLE_DELETION_STARTED, ON_EXTERNAL_FILE_INGESTED,
ON_BACKGROUND_ERROR, ON_STALL_CONDITIONS_CHANGED, ON_FILE_READ_FINISH, ON_FILE_WRITE_FINISH,
ON_FILE_FLUSH_FINISH, ON_FILE_SYNC_FINISH, ON_FILE_RANGE_SYNC_FINISH,
ON_FILE_TRUNCATE_FINISH, ON_FILE_CLOSE_FINISH, SHOULD_BE_NOTIFIED_ON_FILE_IO,
ON_ERROR_RECOVERY_BEGIN, ON_ERROR_RECOVERY_COMPLETED);
}
/**
* Creates an Event Listener that will
* receive only certain callbacks from C++.
*
* @param enabledEventCallbacks callbacks to enable in Java.
*/
protected AbstractEventListener(final EnabledEventCallback... enabledEventCallbacks) {
super(packToLong(enabledEventCallbacks));
}
/**
* Pack EnabledEventCallbacks to a long.
*
* @param enabledEventCallbacks the flags
*
* @return a long
*/
private static long packToLong(final EnabledEventCallback... enabledEventCallbacks) {
long l = 0;
for (int i = 0; i < enabledEventCallbacks.length; i++) {
l |= 1 << enabledEventCallbacks[i].getValue();
}
return l;
}
@Override
public void onFlushCompleted(final RocksDB db, final FlushJobInfo flushJobInfo) {
// no-op
}
/**
* Called from JNI, proxy for
* {@link #onFlushCompleted(RocksDB, FlushJobInfo)}.
*
* @param dbHandle native handle of the database
* @param flushJobInfo the flush job info
*/
private void onFlushCompletedProxy(final long dbHandle, final FlushJobInfo flushJobInfo) {
final RocksDB db = new RocksDB(dbHandle);
db.disOwnNativeHandle(); // we don't own this!
onFlushCompleted(db, flushJobInfo);
}
@Override
public void onFlushBegin(final RocksDB db, final FlushJobInfo flushJobInfo) {
// no-op
}
/**
* Called from JNI, proxy for
* {@link #onFlushBegin(RocksDB, FlushJobInfo)}.
*
* @param dbHandle native handle of the database
* @param flushJobInfo the flush job info
*/
private void onFlushBeginProxy(final long dbHandle, final FlushJobInfo flushJobInfo) {
final RocksDB db = new RocksDB(dbHandle);
db.disOwnNativeHandle(); // we don't own this!
onFlushBegin(db, flushJobInfo);
}
@Override
public void onTableFileDeleted(final TableFileDeletionInfo tableFileDeletionInfo) {
// no-op
}
@Override
public void onCompactionBegin(final RocksDB db, final CompactionJobInfo compactionJobInfo) {
// no-op
}
/**
* Called from JNI, proxy for
* {@link #onCompactionBegin(RocksDB, CompactionJobInfo)}.
*
* @param dbHandle native handle of the database
* @param compactionJobInfo the flush job info
*/
private void onCompactionBeginProxy(
final long dbHandle, final CompactionJobInfo compactionJobInfo) {
final RocksDB db = new RocksDB(dbHandle);
db.disOwnNativeHandle(); // we don't own this!
onCompactionBegin(db, compactionJobInfo);
}
@Override
public void onCompactionCompleted(final RocksDB db, final CompactionJobInfo compactionJobInfo) {
// no-op
}
/**
* Called from JNI, proxy for
* {@link #onCompactionCompleted(RocksDB, CompactionJobInfo)}.
*
* @param dbHandle native handle of the database
* @param compactionJobInfo the flush job info
*/
private void onCompactionCompletedProxy(
final long dbHandle, final CompactionJobInfo compactionJobInfo) {
final RocksDB db = new RocksDB(dbHandle);
db.disOwnNativeHandle(); // we don't own this!
onCompactionCompleted(db, compactionJobInfo);
}
@Override
public void onTableFileCreated(final TableFileCreationInfo tableFileCreationInfo) {
// no-op
}
@Override
public void onTableFileCreationStarted(
final TableFileCreationBriefInfo tableFileCreationBriefInfo) {
// no-op
}
@Override
public void onMemTableSealed(final MemTableInfo memTableInfo) {
// no-op
}
@Override
public void onColumnFamilyHandleDeletionStarted(final ColumnFamilyHandle columnFamilyHandle) {
// no-op
}
@Override
public void onExternalFileIngested(
final RocksDB db, final ExternalFileIngestionInfo externalFileIngestionInfo) {
// no-op
}
/**
* Called from JNI, proxy for
* {@link #onExternalFileIngested(RocksDB, ExternalFileIngestionInfo)}.
*
* @param dbHandle native handle of the database
* @param externalFileIngestionInfo the flush job info
*/
private void onExternalFileIngestedProxy(
final long dbHandle, final ExternalFileIngestionInfo externalFileIngestionInfo) {
final RocksDB db = new RocksDB(dbHandle);
db.disOwnNativeHandle(); // we don't own this!
onExternalFileIngested(db, externalFileIngestionInfo);
}
@Override
public void onBackgroundError(
final BackgroundErrorReason backgroundErrorReason, final Status backgroundError) {
// no-op
}
/**
* Called from JNI, proxy for
* {@link #onBackgroundError(BackgroundErrorReason, Status)}.
*
* @param reasonByte byte value representing error reason
* @param backgroundError status with error code
*/
private void onBackgroundErrorProxy(final byte reasonByte, final Status backgroundError) {
onBackgroundError(BackgroundErrorReason.fromValue(reasonByte), backgroundError);
}
@Override
public void onStallConditionsChanged(final WriteStallInfo writeStallInfo) {
// no-op
}
@Override
public void onFileReadFinish(final FileOperationInfo fileOperationInfo) {
// no-op
}
@Override
public void onFileWriteFinish(final FileOperationInfo fileOperationInfo) {
// no-op
}
@Override
public void OnFileFlushFinish(final FileOperationInfo fileOperationInfo) {
// no-op
}
@Override
public void OnFileSyncFinish(final FileOperationInfo fileOperationInfo) {
// no-op
}
@Override
public void OnFileRangeSyncFinish(final FileOperationInfo fileOperationInfo) {
// no-op
}
@Override
public void OnFileTruncateFinish(final FileOperationInfo fileOperationInfo) {
// no-op
}
@Override
public void OnFileCloseFinish(final FileOperationInfo fileOperationInfo) {
// no-op
}
@Override
public boolean shouldBeNotifiedOnFileIO() {
return false;
}
@Override
public boolean onErrorRecoveryBegin(
final BackgroundErrorReason backgroundErrorReason, final Status backgroundError) {
return true;
}
/**
* Called from JNI, proxy for
* {@link #onErrorRecoveryBegin(BackgroundErrorReason, Status)}.
*
* @param reasonByte byte value representing error reason
* @param backgroundError status with error code
*/
private boolean onErrorRecoveryBeginProxy(final byte reasonByte, final Status backgroundError) {
return onErrorRecoveryBegin(BackgroundErrorReason.fromValue(reasonByte), backgroundError);
}
@Override
public void onErrorRecoveryCompleted(final Status oldBackgroundError) {
// no-op
}
@Override
protected long initializeNative(final long... nativeParameterHandles) {
return createNewEventListener(nativeParameterHandles[0]);
}
/**
* Deletes underlying C++ native callback object pointer
*/
@Override
protected void disposeInternal() {
disposeInternal(nativeHandle_);
}
private native long createNewEventListener(final long enabledEventCallbackValues);
private native void disposeInternal(final long handle);
}

46
java/src/main/java/org/rocksdb/BackgroundErrorReason.java Normal file
View File

@ -0,0 +1,46 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
public enum BackgroundErrorReason {
FLUSH((byte) 0x0),
COMPACTION((byte) 0x1),
WRITE_CALLBACK((byte) 0x2),
MEMTABLE((byte) 0x3);
private final byte value;
BackgroundErrorReason(final byte value) {
this.value = value;
}
/**
* Get the internal representation.
*
* @return the internal representation
*/
byte getValue() {
return value;
}
/**
* Get the BackgroundErrorReason from the internal representation value.
*
* @return the background error reason.
*
* @throws IllegalArgumentException if the value is unknown.
*/
static BackgroundErrorReason fromValue(final byte value) {
for (final BackgroundErrorReason backgroundErrorReason : BackgroundErrorReason.values()) {
if (backgroundErrorReason.value == value) {
return backgroundErrorReason;
}
}
throw new IllegalArgumentException(
"Illegal value provided for BackgroundErrorReason: " + value);
}
}

22
java/src/main/java/org/rocksdb/ColumnFamilyHandle.java
View File

@ -24,6 +24,28 @@ public class ColumnFamilyHandle extends RocksObject {
this.rocksDB_ = rocksDB;
}
/**
* Constructor called only from JNI.
*
* NOTE: we are producing an additional Java Object here to represent the underlying native C++
* ColumnFamilyHandle object. The underlying object is not owned by ourselves. The Java API user
* likely already had a ColumnFamilyHandle Java object which owns the underlying C++ object, as
* they will have been presented it when they opened the database or added a Column Family.
*
*
* TODO(AR) - Potentially a better design would be to cache the active Java Column Family Objects
* in RocksDB, and return the same Java Object instead of instantiating a new one here. This could
* also help us to improve the Java API semantics for Java users. See for example
* https://github.com/facebook/rocksdb/issues/2687.
*
* @param nativeHandle native handle to the column family.
*/
ColumnFamilyHandle(final long nativeHandle) {
super(nativeHandle);
rocksDB_ = null;
disOwnNativeHandle();
}
/**
* Gets the name of the Column Family.
*

2
java/src/main/java/org/rocksdb/CompactionJobInfo.java
View File

@ -20,6 +20,8 @@ public class CompactionJobInfo extends RocksObject {
*/
private CompactionJobInfo(final long nativeHandle) {
super(nativeHandle);
// We do not own the native object!
disOwnNativeHandle();
}
/**

41
java/src/main/java/org/rocksdb/DBOptions.java
View File

@ -884,32 +884,18 @@ public class DBOptions extends RocksObject
return strictBytesPerSync(nativeHandle_);
}
//TODO(AR) NOW
// @Override
// public DBOptions setListeners(final List<EventListener> listeners) {
// assert(isOwningHandle());
// final long[] eventListenerHandlers = new long[listeners.size()];
// for (int i = 0; i < eventListenerHandlers.length; i++) {
// eventListenerHandlers[i] = listeners.get(i).nativeHandle_;
// }
// setEventListeners(nativeHandle_, eventListenerHandlers);
// return this;
// }
//
// @Override
// public Collection<EventListener> listeners() {
// assert(isOwningHandle());
// final long[] eventListenerHandlers = listeners(nativeHandle_);
// if (eventListenerHandlers == null || eventListenerHandlers.length == 0) {
// return Collections.emptyList();
// }
//
// final List<EventListener> eventListeners = new ArrayList<>();
// for (final long eventListenerHandle : eventListenerHandlers) {
// eventListeners.add(new EventListener(eventListenerHandle)); //TODO(AR) check ownership is set to false!
// }
// return eventListeners;
// }
@Override
public DBOptions setListeners(final List<AbstractEventListener> listeners) {
assert (isOwningHandle());
setEventListeners(nativeHandle_, RocksCallbackObject.toNativeHandleList(listeners));
return this;
}
@Override
public List<AbstractEventListener> listeners() {
assert (isOwningHandle());
return Arrays.asList(eventListeners(nativeHandle_));
}
@Override
public DBOptions setEnableThreadTracking(final boolean enableThreadTracking) {
@ -1459,6 +1445,9 @@ public class DBOptions extends RocksObject
final long handle, final boolean strictBytesPerSync);
private native boolean strictBytesPerSync(
final long handle);
private static native void setEventListeners(
final long handle, final long[] eventListenerHandles);
private static native AbstractEventListener[] eventListeners(final long handle);
private native void setEnableThreadTracking(long handle,
boolean enableThreadTracking);
private native boolean enableThreadTracking(long handle);

43
java/src/main/java/org/rocksdb/DBOptionsInterface.java
View File

@ -1055,24 +1055,31 @@ public interface DBOptionsInterface<T extends DBOptionsInterface<T>> {
*/
boolean useAdaptiveMutex();
//TODO(AR) NOW
// /**
// * Sets the {@link EventListener}s whose callback functions
// * will be called when specific RocksDB event happens.
// *
// * @param listeners the listeners who should be notified on various events.
// *
// * @return the instance of the current object.
// */
// T setListeners(final List<EventListener> listeners);
//
// /**
// * Gets the {@link EventListener}s whose callback functions
// * will be called when specific RocksDB event happens.
// *
// * @return a collection of Event listeners.
// */
// Collection<EventListener> listeners();
/**
* Sets the {@link EventListener}s whose callback functions
* will be called when specific RocksDB event happens.
*
* Note: the RocksJava API currently only supports EventListeners implemented in Java.
* It could be extended in future to also support adding/removing EventListeners implemented in
* C++.
*
* @param listeners the listeners who should be notified on various events.
*
* @return the instance of the current object.
*/
T setListeners(final List<AbstractEventListener> listeners);
/**
* Sets the {@link EventListener}s whose callback functions
* will be called when specific RocksDB event happens.
*
* Note: the RocksJava API currently only supports EventListeners implemented in Java.
* It could be extended in future to also support adding/removing EventListeners implemented in
* C++.
*
* @return the instance of the current object.
*/
List<AbstractEventListener> listeners();
/**
* If true, then the status of the threads involved in this DB will

332
java/src/main/java/org/rocksdb/EventListener.java Normal file
View File

@ -0,0 +1,332 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.List;
/**
* EventListener class contains a set of callback functions that will
* be called when specific RocksDB event happens such as flush. It can
* be used as a building block for developing custom features such as
* stats-collector or external compaction algorithm.
*
* Note that callback functions should not run for an extended period of
* time before the function returns, otherwise RocksDB may be blocked.
* For example, it is not suggested to do
* {@link RocksDB#compactFiles(CompactionOptions, ColumnFamilyHandle, List, int, int,
* CompactionJobInfo)} (as it may run for a long while) or issue many of
* {@link RocksDB#put(ColumnFamilyHandle, WriteOptions, byte[], byte[])}
* (as Put may be blocked in certain cases) in the same thread in the
* EventListener callback.
*
* However, doing
* {@link RocksDB#compactFiles(CompactionOptions, ColumnFamilyHandle, List, int, int,
* CompactionJobInfo)} and {@link RocksDB#put(ColumnFamilyHandle, WriteOptions, byte[], byte[])} in
* another thread is considered safe.
*
* [Threading] All EventListener callback will be called using the
* actual thread that involves in that specific event. For example, it
* is the RocksDB background flush thread that does the actual flush to
* call {@link #onFlushCompleted(RocksDB, FlushJobInfo)}.
*
* [Locking] All EventListener callbacks are designed to be called without
* the current thread holding any DB mutex. This is to prevent potential
* deadlock and performance issue when using EventListener callback
* in a complex way.
*/
public interface EventListener {
/**
* A callback function to RocksDB which will be called before a
* RocksDB starts to flush memtables.
*
* Note that the this function must be implemented in a way such that
* it should not run for an extended period of time before the function
* returns. Otherwise, RocksDB may be blocked.
*
* @param db the database
* @param flushJobInfo the flush job info, contains data copied from
* respective native structure.
*/
void onFlushBegin(final RocksDB db, final FlushJobInfo flushJobInfo);
/**
* callback function to RocksDB which will be called whenever a
* registered RocksDB flushes a file.
*
* Note that the this function must be implemented in a way such that
* it should not run for an extended period of time before the function
* returns. Otherwise, RocksDB may be blocked.
*
* @param db the database
* @param flushJobInfo the flush job info, contains data copied from
* respective native structure.
*/
void onFlushCompleted(final RocksDB db, final FlushJobInfo flushJobInfo);
/**
* A callback function for RocksDB which will be called whenever
* a SST file is deleted. Different from
* {@link #onCompactionCompleted(RocksDB, CompactionJobInfo)} and
* {@link #onFlushCompleted(RocksDB, FlushJobInfo)},
* this callback is designed for external logging
* service and thus only provide string parameters instead
* of a pointer to DB. Applications that build logic basic based
* on file creations and deletions is suggested to implement
* {@link #onFlushCompleted(RocksDB, FlushJobInfo)} and
* {@link #onCompactionCompleted(RocksDB, CompactionJobInfo)}.
*
* Note that if applications would like to use the passed reference
* outside this function call, they should make copies from the
* returned value.
*
* @param tableFileDeletionInfo the table file deletion info,
* contains data copied from respective native structure.
*/
void onTableFileDeleted(final TableFileDeletionInfo tableFileDeletionInfo);
/**
* A callback function to RocksDB which will be called before a
* RocksDB starts to compact. The default implementation is
* no-op.
*
* Note that the this function must be implemented in a way such that
* it should not run for an extended period of time before the function
* returns. Otherwise, RocksDB may be blocked.
*
* @param db a pointer to the rocksdb instance which just compacted
* a file.
* @param compactionJobInfo a reference to a native CompactionJobInfo struct,
* which is released after this function is returned, and must be copied
* if it is needed outside of this function.
*/
void onCompactionBegin(final RocksDB db, final CompactionJobInfo compactionJobInfo);
/**
* A callback function for RocksDB which will be called whenever
* a registered RocksDB compacts a file. The default implementation
* is a no-op.
*
* Note that this function must be implemented in a way such that
* it should not run for an extended period of time before the function
* returns. Otherwise, RocksDB may be blocked.
*
* @param db a pointer to the rocksdb instance which just compacted
* a file.
* @param compactionJobInfo a reference to a native CompactionJobInfo struct,
* which is released after this function is returned, and must be copied
* if it is needed outside of this function.
*/
void onCompactionCompleted(final RocksDB db, final CompactionJobInfo compactionJobInfo);
/**
* A callback function for RocksDB which will be called whenever
* a SST file is created. Different from OnCompactionCompleted and
* OnFlushCompleted, this callback is designed for external logging
* service and thus only provide string parameters instead
* of a pointer to DB. Applications that build logic basic based
* on file creations and deletions is suggested to implement
* OnFlushCompleted and OnCompactionCompleted.
*
* Historically it will only be called if the file is successfully created.
* Now it will also be called on failure case. User can check info.status
* to see if it succeeded or not.
*
* Note that if applications would like to use the passed reference
* outside this function call, they should make copies from these
* returned value.
*
* @param tableFileCreationInfo the table file creation info,
* contains data copied from respective native structure.
*/
void onTableFileCreated(final TableFileCreationInfo tableFileCreationInfo);
/**
* A callback function for RocksDB which will be called before
* a SST file is being created. It will follow by OnTableFileCreated after
* the creation finishes.
*
* Note that if applications would like to use the passed reference
* outside this function call, they should make copies from these
* returned value.
*
* @param tableFileCreationBriefInfo the table file creation brief info,
* contains data copied from respective native structure.
*/
void onTableFileCreationStarted(final TableFileCreationBriefInfo tableFileCreationBriefInfo);
/**
* A callback function for RocksDB which will be called before
* a memtable is made immutable.
*
* Note that the this function must be implemented in a way such that
* it should not run for an extended period of time before the function
* returns. Otherwise, RocksDB may be blocked.
*
* Note that if applications would like to use the passed reference
* outside this function call, they should make copies from these
* returned value.
*
* @param memTableInfo the mem table info, contains data
* copied from respective native structure.
*/
void onMemTableSealed(final MemTableInfo memTableInfo);
/**
* A callback function for RocksDB which will be called before
* a column family handle is deleted.
*
* Note that the this function must be implemented in a way such that
* it should not run for an extended period of time before the function
* returns. Otherwise, RocksDB may be blocked.
*
* @param columnFamilyHandle is a pointer to the column family handle to be
* deleted which will become a dangling pointer after the deletion.
*/
void onColumnFamilyHandleDeletionStarted(final ColumnFamilyHandle columnFamilyHandle);
/**
* A callback function for RocksDB which will be called after an external
* file is ingested using IngestExternalFile.
*
* Note that the this function will run on the same thread as
* IngestExternalFile(), if this function is blocked, IngestExternalFile()
* will be blocked from finishing.
*
* @param db the database
* @param externalFileIngestionInfo the external file ingestion info,
* contains data copied from respective native structure.
*/
void onExternalFileIngested(
final RocksDB db, final ExternalFileIngestionInfo externalFileIngestionInfo);
/**
* A callback function for RocksDB which will be called before setting the
* background error status to a non-OK value. The new background error status
* is provided in `bg_error` and can be modified by the callback. E.g., a
* callback can suppress errors by resetting it to Status::OK(), thus
* preventing the database from entering read-only mode. We do not provide any
* guarantee when failed flushes/compactions will be rescheduled if the user
* suppresses an error.
*
* Note that this function can run on the same threads as flush, compaction,
* and user writes. So, it is extremely important not to perform heavy
* computations or blocking calls in this function.
*
* @param backgroundErrorReason background error reason code
* @param backgroundError background error codes
*/
void onBackgroundError(
final BackgroundErrorReason backgroundErrorReason, final Status backgroundError);
/**
* A callback function for RocksDB which will be called whenever a change
* of superversion triggers a change of the stall conditions.
*
* Note that the this function must be implemented in a way such that
* it should not run for an extended period of time before the function
* returns. Otherwise, RocksDB may be blocked.
*
* @param writeStallInfo write stall info,
* contains data copied from respective native structure.
*/
void onStallConditionsChanged(final WriteStallInfo writeStallInfo);
/**
* A callback function for RocksDB which will be called whenever a file read
* operation finishes.
*
* @param fileOperationInfo file operation info,
* contains data copied from respective native structure.
*/
void onFileReadFinish(final FileOperationInfo fileOperationInfo);
/**
* A callback function for RocksDB which will be called whenever a file write
* operation finishes.
*
* @param fileOperationInfo file operation info,
* contains data copied from respective native structure.
*/
void onFileWriteFinish(final FileOperationInfo fileOperationInfo);
/**
* A callback function for RocksDB which will be called whenever a file flush
* operation finishes.
*
* @param fileOperationInfo file operation info,
* contains data copied from respective native structure.
*/
void OnFileFlushFinish(final FileOperationInfo fileOperationInfo);
/**
* A callback function for RocksDB which will be called whenever a file sync
* operation finishes.
*
* @param fileOperationInfo file operation info,
* contains data copied from respective native structure.
*/
void OnFileSyncFinish(final FileOperationInfo fileOperationInfo);
/**
* A callback function for RocksDB which will be called whenever a file
* rangeSync operation finishes.
*
* @param fileOperationInfo file operation info,
* contains data copied from respective native structure.
*/
void OnFileRangeSyncFinish(final FileOperationInfo fileOperationInfo);
/**
* A callback function for RocksDB which will be called whenever a file
* truncate operation finishes.
*
* @param fileOperationInfo file operation info,
* contains data copied from respective native structure.
*/
void OnFileTruncateFinish(final FileOperationInfo fileOperationInfo);
/**
* A callback function for RocksDB which will be called whenever a file close
* operation finishes.
*
* @param fileOperationInfo file operation info,
* contains data copied from respective native structure.
*/
void OnFileCloseFinish(final FileOperationInfo fileOperationInfo);
/**
* If true, the {@link #onFileReadFinish(FileOperationInfo)}
* and {@link #onFileWriteFinish(FileOperationInfo)} will be called. If
* false, then they won't be called.
*
* Default: false
*/
boolean shouldBeNotifiedOnFileIO();
/**
* A callback function for RocksDB which will be called just before
* starting the automatic recovery process for recoverable background
* errors, such as NoSpace(). The callback can suppress the automatic
* recovery by setting returning false. The database will then
* have to be transitioned out of read-only mode by calling
* RocksDB#resume().
*
* @param backgroundErrorReason background error reason code
* @param backgroundError background error codes
*/
boolean onErrorRecoveryBegin(
final BackgroundErrorReason backgroundErrorReason, final Status backgroundError);
/**
* A callback function for RocksDB which will be called once the database
* is recovered from read-only mode after an error. When this is called, it
* means normal writes to the database can be issued and the user can
* initiate any further recovery actions needed
*
* @param oldBackgroundError old background error codes
*/
void onErrorRecoveryCompleted(final Status oldBackgroundError);
}

103
java/src/main/java/org/rocksdb/ExternalFileIngestionInfo.java Normal file
View File

@ -0,0 +1,103 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.Objects;
public class ExternalFileIngestionInfo {
private final String columnFamilyName;
private final String externalFilePath;
private final String internalFilePath;
private final long globalSeqno;
private final TableProperties tableProperties;
/**
* Access is package private as this will only be constructed from
* C++ via JNI and for testing.
*/
ExternalFileIngestionInfo(final String columnFamilyName, final String externalFilePath,
final String internalFilePath, final long globalSeqno,
final TableProperties tableProperties) {
this.columnFamilyName = columnFamilyName;
this.externalFilePath = externalFilePath;
this.internalFilePath = internalFilePath;
this.globalSeqno = globalSeqno;
this.tableProperties = tableProperties;
}
/**
* Get the name of the column family.
*
* @return the name of the column family.
*/
public String getColumnFamilyName() {
return columnFamilyName;
}
/**
* Get the path of the file outside the DB.
*
* @return the path of the file outside the DB.
*/
public String getExternalFilePath() {
return externalFilePath;
}
/**
* Get the path of the file inside the DB.
*
* @return the path of the file inside the DB.
*/
public String getInternalFilePath() {
return internalFilePath;
}
/**
* Get the global sequence number assigned to keys in this file.
*
* @return the global sequence number.
*/
public long getGlobalSeqno() {
return globalSeqno;
}
/**
* Get the Table properties of the table being flushed.
*
* @return the table properties.
*/
public TableProperties getTableProperties() {
return tableProperties;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
ExternalFileIngestionInfo that = (ExternalFileIngestionInfo) o;
return globalSeqno == that.globalSeqno
&& Objects.equals(columnFamilyName, that.columnFamilyName)
&& Objects.equals(externalFilePath, that.externalFilePath)
&& Objects.equals(internalFilePath, that.internalFilePath)
&& Objects.equals(tableProperties, that.tableProperties);
}
@Override
public int hashCode() {
return Objects.hash(
columnFamilyName, externalFilePath, internalFilePath, globalSeqno, tableProperties);
}
@Override
public String toString() {
return "ExternalFileIngestionInfo{"
+ "columnFamilyName='" + columnFamilyName + '\'' + ", externalFilePath='" + externalFilePath
+ '\'' + ", internalFilePath='" + internalFilePath + '\'' + ", globalSeqno=" + globalSeqno
+ ", tableProperties=" + tableProperties + '}';
}
}

112
java/src/main/java/org/rocksdb/FileOperationInfo.java Normal file
View File

@ -0,0 +1,112 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.Objects;
/**
* Java representation of FileOperationInfo struct from include/rocksdb/listener.h
*/
public class FileOperationInfo {
private final String path;
private final long offset;
private final long length;
private final long startTimestamp;
private final long duration;
private final Status status;
/**
* Access is private as this will only be constructed from
* C++ via JNI.
*/
FileOperationInfo(final String path, final long offset, final long length,
final long startTimestamp, final long duration, final Status status) {
this.path = path;
this.offset = offset;
this.length = length;
this.startTimestamp = startTimestamp;
this.duration = duration;
this.status = status;
}
/**
* Get the file path.
*
* @return the file path.
*/
public String getPath() {
return path;
}
/**
* Get the offset.
*
* @return the offset.
*/
public long getOffset() {
return offset;
}
/**
* Get the length.
*
* @return the length.
*/
public long getLength() {
return length;
}
/**
* Get the start timestamp (in nanoseconds).
*
* @return the start timestamp.
*/
public long getStartTimestamp() {
return startTimestamp;
}
/**
* Get the operation duration (in nanoseconds).
*
* @return the operation duration.
*/
public long getDuration() {
return duration;
}
/**
* Get the status.
*
* @return the status.
*/
public Status getStatus() {
return status;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
FileOperationInfo that = (FileOperationInfo) o;
return offset == that.offset && length == that.length && startTimestamp == that.startTimestamp
&& duration == that.duration && Objects.equals(path, that.path)
&& Objects.equals(status, that.status);
}
@Override
public int hashCode() {
return Objects.hash(path, offset, length, startTimestamp, duration, status);
}
@Override
public String toString() {
return "FileOperationInfo{"
+ "path='" + path + '\'' + ", offset=" + offset + ", length=" + length + ", startTimestamp="
+ startTimestamp + ", duration=" + duration + ", status=" + status + '}';
}
}

186
java/src/main/java/org/rocksdb/FlushJobInfo.java Normal file
View File

@ -0,0 +1,186 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.Objects;
public class FlushJobInfo {
private final long columnFamilyId;
private final String columnFamilyName;
private final String filePath;
private final long threadId;
private final int jobId;
private final boolean triggeredWritesSlowdown;
private final boolean triggeredWritesStop;
private final long smallestSeqno;
private final long largestSeqno;
private final TableProperties tableProperties;
private final FlushReason flushReason;
/**
* Access is package private as this will only be constructed from
* C++ via JNI and for testing.
*/
FlushJobInfo(final long columnFamilyId, final String columnFamilyName, final String filePath,
final long threadId, final int jobId, final boolean triggeredWritesSlowdown,
final boolean triggeredWritesStop, final long smallestSeqno, final long largestSeqno,
final TableProperties tableProperties, final byte flushReasonValue) {
this.columnFamilyId = columnFamilyId;
this.columnFamilyName = columnFamilyName;
this.filePath = filePath;
this.threadId = threadId;
this.jobId = jobId;
this.triggeredWritesSlowdown = triggeredWritesSlowdown;
this.triggeredWritesStop = triggeredWritesStop;
this.smallestSeqno = smallestSeqno;
this.largestSeqno = largestSeqno;
this.tableProperties = tableProperties;
this.flushReason = FlushReason.fromValue(flushReasonValue);
}
/**
* Get the id of the column family.
*
* @return the id of the column family
*/
public long getColumnFamilyId() {
return columnFamilyId;
}
/**
* Get the name of the column family.
*
* @return the name of the column family
*/
public String getColumnFamilyName() {
return columnFamilyName;
}
/**
* Get the path to the newly created file.
*
* @return the path to the newly created file
*/
public String getFilePath() {
return filePath;
}
/**
* Get the id of the thread that completed this flush job.
*
* @return the id of the thread that completed this flush job
*/
public long getThreadId() {
return threadId;
}
/**
* Get the job id, which is unique in the same thread.
*
* @return the job id
*/
public int getJobId() {
return jobId;
}
/**
* Determine if rocksdb is currently slowing-down all writes to prevent
* creating too many Level 0 files as compaction seems not able to
* catch up the write request speed.
*
* This indicates that there are too many files in Level 0.
*
* @return true if rocksdb is currently slowing-down all writes,
* false otherwise
*/
public boolean isTriggeredWritesSlowdown() {
return triggeredWritesSlowdown;
}
/**
* Determine if rocksdb is currently blocking any writes to prevent
* creating more L0 files.
*
* This indicates that there are too many files in level 0.
* Compactions should try to compact L0 files down to lower levels as soon
* as possible.
*
* @return true if rocksdb is currently blocking any writes, false otherwise
*/
public boolean isTriggeredWritesStop() {
return triggeredWritesStop;
}
/**
* Get the smallest sequence number in the newly created file.
*
* @return the smallest sequence number
*/
public long getSmallestSeqno() {
return smallestSeqno;
}
/**
* Get the largest sequence number in the newly created file.
*
* @return the largest sequence number
*/
public long getLargestSeqno() {
return largestSeqno;
}
/**
* Get the Table properties of the table being flushed.
*
* @return the Table properties of the table being flushed
*/
public TableProperties getTableProperties() {
return tableProperties;
}
/**
* Get the reason for initiating the flush.
*
* @return the reason for initiating the flush.
*/
public FlushReason getFlushReason() {
return flushReason;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
FlushJobInfo that = (FlushJobInfo) o;
return columnFamilyId == that.columnFamilyId && threadId == that.threadId && jobId == that.jobId
&& triggeredWritesSlowdown == that.triggeredWritesSlowdown
&& triggeredWritesStop == that.triggeredWritesStop && smallestSeqno == that.smallestSeqno
&& largestSeqno == that.largestSeqno
&& Objects.equals(columnFamilyName, that.columnFamilyName)
&& Objects.equals(filePath, that.filePath)
&& Objects.equals(tableProperties, that.tableProperties) && flushReason == that.flushReason;
}
@Override
public int hashCode() {
return Objects.hash(columnFamilyId, columnFamilyName, filePath, threadId, jobId,
triggeredWritesSlowdown, triggeredWritesStop, smallestSeqno, largestSeqno, tableProperties,
flushReason);
}
@Override
public String toString() {
return "FlushJobInfo{"
+ "columnFamilyId=" + columnFamilyId + ", columnFamilyName='" + columnFamilyName + '\''
+ ", filePath='" + filePath + '\'' + ", threadId=" + threadId + ", jobId=" + jobId
+ ", triggeredWritesSlowdown=" + triggeredWritesSlowdown
+ ", triggeredWritesStop=" + triggeredWritesStop + ", smallestSeqno=" + smallestSeqno
+ ", largestSeqno=" + largestSeqno + ", tableProperties=" + tableProperties
+ ", flushReason=" + flushReason + '}';
}
}

53
java/src/main/java/org/rocksdb/FlushReason.java Normal file
View File

@ -0,0 +1,53 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
public enum FlushReason {
OTHERS((byte) 0x00),
GET_LIVE_FILES((byte) 0x01),
SHUTDOWN((byte) 0x02),
EXTERNAL_FILE_INGESTION((byte) 0x03),
MANUAL_COMPACTION((byte) 0x04),
WRITE_BUFFER_MANAGER((byte) 0x05),
WRITE_BUFFER_FULL((byte) 0x06),
TEST((byte) 0x07),
DELETE_FILES((byte) 0x08),
AUTO_COMPACTION((byte) 0x09),
MANUAL_FLUSH((byte) 0x0a),
ERROR_RECOVERY((byte) 0xb);
private final byte value;
FlushReason(final byte value) {
this.value = value;
}
/**
* Get the internal representation.
*
* @return the internal representation
*/
byte getValue() {
return value;
}
/**
* Get the FlushReason from the internal representation value.
*
* @return the flush reason.
*
* @throws IllegalArgumentException if the value is unknown.
*/
static FlushReason fromValue(final byte value) {
for (final FlushReason flushReason : FlushReason.values()) {
if (flushReason.value == value) {
return flushReason;
}
}
throw new IllegalArgumentException("Illegal value provided for FlushReason: " + value);
}
}

103
java/src/main/java/org/rocksdb/MemTableInfo.java Normal file
View File

@ -0,0 +1,103 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.Objects;
public class MemTableInfo {
private final String columnFamilyName;
private final long firstSeqno;
private final long earliestSeqno;
private final long numEntries;
private final long numDeletes;
/**
* Access is package private as this will only be constructed from
* C++ via JNI and for testing.
*/
MemTableInfo(final String columnFamilyName, final long firstSeqno, final long earliestSeqno,
final long numEntries, final long numDeletes) {
this.columnFamilyName = columnFamilyName;
this.firstSeqno = firstSeqno;
this.earliestSeqno = earliestSeqno;
this.numEntries = numEntries;
this.numDeletes = numDeletes;
}
/**
* Get the name of the column family to which memtable belongs.
*
* @return the name of the column family.
*/
public String getColumnFamilyName() {
return columnFamilyName;
}
/**
* Get the Sequence number of the first element that was inserted into the
* memtable.
*
* @return the sequence number of the first inserted element.
*/
public long getFirstSeqno() {
return firstSeqno;
}
/**
* Get the Sequence number that is guaranteed to be smaller than or equal
* to the sequence number of any key that could be inserted into this
* memtable. It can then be assumed that any write with a larger(or equal)
* sequence number will be present in this memtable or a later memtable.
*
* @return the earliest sequence number.
*/
public long getEarliestSeqno() {
return earliestSeqno;
}
/**
* Get the total number of entries in memtable.
*
* @return the total number of entries.
*/
public long getNumEntries() {
return numEntries;
}
/**
* Get the total number of deletes in memtable.
*
* @return the total number of deletes.
*/
public long getNumDeletes() {
return numDeletes;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
MemTableInfo that = (MemTableInfo) o;
return firstSeqno == that.firstSeqno && earliestSeqno == that.earliestSeqno
&& numEntries == that.numEntries && numDeletes == that.numDeletes
&& Objects.equals(columnFamilyName, that.columnFamilyName);
}
@Override
public int hashCode() {
return Objects.hash(columnFamilyName, firstSeqno, earliestSeqno, numEntries, numDeletes);
}
@Override
public String toString() {
return "MemTableInfo{"
+ "columnFamilyName='" + columnFamilyName + '\'' + ", firstSeqno=" + firstSeqno
+ ", earliestSeqno=" + earliestSeqno + ", numEntries=" + numEntries
+ ", numDeletes=" + numDeletes + '}';
}
}

16
java/src/main/java/org/rocksdb/Options.java
View File

@ -970,6 +970,19 @@ public class Options extends RocksObject
return strictBytesPerSync(nativeHandle_);
}
@Override
public Options setListeners(final List<AbstractEventListener> listeners) {
assert (isOwningHandle());
setEventListeners(nativeHandle_, RocksCallbackObject.toNativeHandleList(listeners));
return this;
}
@Override
public List<AbstractEventListener> listeners() {
assert (isOwningHandle());
return Arrays.asList(eventListeners(nativeHandle_));
}
@Override
public Options setEnableThreadTracking(final boolean enableThreadTracking) {
assert(isOwningHandle());
@ -2151,6 +2164,9 @@ public class Options extends RocksObject
final long handle, final boolean strictBytesPerSync);
private native boolean strictBytesPerSync(
final long handle);
private static native void setEventListeners(
final long handle, final long[] eventListenerHandles);
private static native AbstractEventListener[] eventListeners(final long handle);
private native void setEnableThreadTracking(long handle,
boolean enableThreadTracking);
private native boolean enableThreadTracking(long handle);

23
java/src/main/java/org/rocksdb/RocksCallbackObject.java
View File

@ -5,6 +5,8 @@
package org.rocksdb;
import java.util.List;
/**
* RocksCallbackObject is similar to {@link RocksObject} but varies
* in its construction as it is designed for Java objects which have functions
@ -26,6 +28,27 @@ public abstract class RocksCallbackObject extends
this.nativeHandle_ = initializeNative(nativeParameterHandles);
}
/**
* Given a list of RocksCallbackObjects, it returns a list
* of the native handles of the underlying objects.
*
* @param objectList the rocks callback objects
*
* @return the native handles
*/
static /* @Nullable */ long[] toNativeHandleList(
/* @Nullable */ final List<? extends RocksCallbackObject> objectList) {
if (objectList == null) {
return null;
}
final int len = objectList.size();
final long[] handleList = new long[len];
for (int i = 0; i < len; i++) {
handleList[i] = objectList.get(i).nativeHandle_;
}
return handleList;
}
/**
* Construct the Native C++ object which will callback
* to our object methods

17
java/src/main/java/org/rocksdb/Status.java
View File

@ -5,6 +5,8 @@
package org.rocksdb;
import java.util.Objects;
/**
* Represents the status returned by a function call in RocksDB.
*
@ -135,4 +137,19 @@ public class Status {
return value;
}
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
Status status = (Status) o;
return code == status.code && subCode == status.subCode && Objects.equals(state, status.state);
}
@Override
public int hashCode() {
return Objects.hash(code, subCode, state);
}
}

107
java/src/main/java/org/rocksdb/TableFileCreationBriefInfo.java Normal file
View File

@ -0,0 +1,107 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.Objects;
public class TableFileCreationBriefInfo {
private final String dbName;
private final String columnFamilyName;
private final String filePath;
private final int jobId;
private final TableFileCreationReason reason;
/**
* Access is private as this will only be constructed from
* C++ via JNI, either directly of via
* {@link TableFileCreationInfo#TableFileCreationInfo(long, TableProperties, Status, String,
* String, String, int, byte)}.
*
* @param dbName the database name
* @param columnFamilyName the column family name
* @param filePath the path to the table file
* @param jobId the job identifier
* @param tableFileCreationReasonValue the reason for creation of the table file
*/
protected TableFileCreationBriefInfo(final String dbName, final String columnFamilyName,
final String filePath, final int jobId, final byte tableFileCreationReasonValue) {
this.dbName = dbName;
this.columnFamilyName = columnFamilyName;
this.filePath = filePath;
this.jobId = jobId;
this.reason = TableFileCreationReason.fromValue(tableFileCreationReasonValue);
}
/**
* Get the name of the database where the file was created.
*
* @return the name of the database.
*/
public String getDbName() {
return dbName;
}
/**
* Get the name of the column family where the file was created.
*
* @return the name of the column family.
*/
public String getColumnFamilyName() {
return columnFamilyName;
}
/**
* Get the path to the created file.
*
* @return the path.
*/
public String getFilePath() {
return filePath;
}
/**
* Get the id of the job (which could be flush or compaction) that
* created the file.
*
* @return the id of the job.
*/
public int getJobId() {
return jobId;
}
/**
* Get the reason for creating the table.
*
* @return the reason for creating the table.
*/
public TableFileCreationReason getReason() {
return reason;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
TableFileCreationBriefInfo that = (TableFileCreationBriefInfo) o;
return jobId == that.jobId && Objects.equals(dbName, that.dbName)
&& Objects.equals(columnFamilyName, that.columnFamilyName)
&& Objects.equals(filePath, that.filePath) && reason == that.reason;
}
@Override
public int hashCode() {
return Objects.hash(dbName, columnFamilyName, filePath, jobId, reason);
}
@Override
public String toString() {
return "TableFileCreationBriefInfo{"
+ "dbName='" + dbName + '\'' + ", columnFamilyName='" + columnFamilyName + '\''
+ ", filePath='" + filePath + '\'' + ", jobId=" + jobId + ", reason=" + reason + '}';
}
}

86
java/src/main/java/org/rocksdb/TableFileCreationInfo.java Normal file
View File

@ -0,0 +1,86 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.Objects;
public class TableFileCreationInfo extends TableFileCreationBriefInfo {
private final long fileSize;
private final TableProperties tableProperties;
private final Status status;
/**
* Access is protected as this will only be constructed from
* C++ via JNI.
*
* @param fileSize the size of the table file
* @param tableProperties the properties of the table file
* @param status the status of the creation operation
* @param dbName the database name
* @param columnFamilyName the column family name
* @param filePath the path to the table file
* @param jobId the job identifier
* @param tableFileCreationReasonValue the reason for creation of the table file
*/
protected TableFileCreationInfo(final long fileSize, final TableProperties tableProperties,
final Status status, final String dbName, final String columnFamilyName,
final String filePath, final int jobId, final byte tableFileCreationReasonValue) {
super(dbName, columnFamilyName, filePath, jobId, tableFileCreationReasonValue);
this.fileSize = fileSize;
this.tableProperties = tableProperties;
this.status = status;
}
/**
* Get the size of the file.
*
* @return the size.
*/
public long getFileSize() {
return fileSize;
}
/**
* Get the detailed properties of the created file.
*
* @return the properties.
*/
public TableProperties getTableProperties() {
return tableProperties;
}
/**
* Get the status indicating whether the creation was successful or not.
*
* @return the status.
*/
public Status getStatus() {
return status;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
TableFileCreationInfo that = (TableFileCreationInfo) o;
return fileSize == that.fileSize && Objects.equals(tableProperties, that.tableProperties)
&& Objects.equals(status, that.status);
}
@Override
public int hashCode() {
return Objects.hash(fileSize, tableProperties, status);
}
@Override
public String toString() {
return "TableFileCreationInfo{"
+ "fileSize=" + fileSize + ", tableProperties=" + tableProperties + ", status=" + status
+ '}';
}
}

46
java/src/main/java/org/rocksdb/TableFileCreationReason.java Normal file
View File

@ -0,0 +1,46 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
public enum TableFileCreationReason {
FLUSH((byte) 0x00),
COMPACTION((byte) 0x01),
RECOVERY((byte) 0x02),
MISC((byte) 0x03);
private final byte value;
TableFileCreationReason(final byte value) {
this.value = value;
}
/**
* Get the internal representation.
*
* @return the internal representation
*/
byte getValue() {
return value;
}
/**
* Get the TableFileCreationReason from the internal representation value.
*
* @return the table file creation reason.
*
* @throws IllegalArgumentException if the value is unknown.
*/
static TableFileCreationReason fromValue(final byte value) {
for (final TableFileCreationReason tableFileCreationReason : TableFileCreationReason.values()) {
if (tableFileCreationReason.value == value) {
return tableFileCreationReason;
}
}
throw new IllegalArgumentException(
"Illegal value provided for TableFileCreationReason: " + value);
}
}

86
java/src/main/java/org/rocksdb/TableFileDeletionInfo.java Normal file
View File

@ -0,0 +1,86 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.Objects;
public class TableFileDeletionInfo {
private final String dbName;
private final String filePath;
private final int jobId;
private final Status status;
/**
* Access is package private as this will only be constructed from
* C++ via JNI and for testing.
*/
TableFileDeletionInfo(
final String dbName, final String filePath, final int jobId, final Status status) {
this.dbName = dbName;
this.filePath = filePath;
this.jobId = jobId;
this.status = status;
}
/**
* Get the name of the database where the file was deleted.
*
* @return the name of the database.
*/
public String getDbName() {
return dbName;
}
/**
* Get the path to the deleted file.
*
* @return the path.
*/
public String getFilePath() {
return filePath;
}
/**
* Get the id of the job which deleted the file.
*
* @return the id of the job.
*/
public int getJobId() {
return jobId;
}
/**
* Get the status indicating whether the deletion was successful or not.
*
* @return the status
*/
public Status getStatus() {
return status;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
TableFileDeletionInfo that = (TableFileDeletionInfo) o;
return jobId == that.jobId && Objects.equals(dbName, that.dbName)
&& Objects.equals(filePath, that.filePath) && Objects.equals(status, that.status);
}
@Override
public int hashCode() {
return Objects.hash(dbName, filePath, jobId, status);
}
@Override
public String toString() {
return "TableFileDeletionInfo{"
+ "dbName='" + dbName + '\'' + ", filePath='" + filePath + '\'' + ", jobId=" + jobId
+ ", status=" + status + '}';
}
}

74
java/src/main/java/org/rocksdb/TableProperties.java
View File

@ -1,7 +1,9 @@
// Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
package org.rocksdb;
import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
/**
* TableProperties contains read-only properties of its associated
@ -39,24 +41,20 @@ public class TableProperties {
private final Map<String, Long> propertiesOffsets;
/**
* Access is private as this will only be constructed from
* C++ via JNI.
* Access is package private as this will only be constructed from
* C++ via JNI and for testing.
*/
private TableProperties(final long dataSize, final long indexSize,
final long indexPartitions, final long topLevelIndexSize,
final long indexKeyIsUserKey, final long indexValueIsDeltaEncoded,
final long filterSize, final long rawKeySize, final long rawValueSize,
final long numDataBlocks, final long numEntries, final long numDeletions,
final long numMergeOperands, final long numRangeDeletions,
final long formatVersion, final long fixedKeyLen,
final long columnFamilyId, final long creationTime,
final long oldestKeyTime, final byte[] columnFamilyName,
final String filterPolicyName, final String comparatorName,
final String mergeOperatorName, final String prefixExtractorName,
final String propertyCollectorsNames, final String compressionName,
final Map<String, String> userCollectedProperties,
final Map<String, String> readableProperties,
final Map<String, Long> propertiesOffsets) {
TableProperties(final long dataSize, final long indexSize, final long indexPartitions,
final long topLevelIndexSize, final long indexKeyIsUserKey,
final long indexValueIsDeltaEncoded, final long filterSize, final long rawKeySize,
final long rawValueSize, final long numDataBlocks, final long numEntries,
final long numDeletions, final long numMergeOperands, final long numRangeDeletions,
final long formatVersion, final long fixedKeyLen, final long columnFamilyId,
final long creationTime, final long oldestKeyTime, final byte[] columnFamilyName,
final String filterPolicyName, final String comparatorName, final String mergeOperatorName,
final String prefixExtractorName, final String propertyCollectorsNames,
final String compressionName, final Map<String, String> userCollectedProperties,
final Map<String, String> readableProperties, final Map<String, Long> propertiesOffsets) {
this.dataSize = dataSize;
this.indexSize = indexSize;
this.indexPartitions = indexPartitions;
@ -363,4 +361,46 @@ public class TableProperties {
public Map<String, Long> getPropertiesOffsets() {
return propertiesOffsets;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
TableProperties that = (TableProperties) o;
return dataSize == that.dataSize && indexSize == that.indexSize
&& indexPartitions == that.indexPartitions && topLevelIndexSize == that.topLevelIndexSize
&& indexKeyIsUserKey == that.indexKeyIsUserKey
&& indexValueIsDeltaEncoded == that.indexValueIsDeltaEncoded
&& filterSize == that.filterSize && rawKeySize == that.rawKeySize
&& rawValueSize == that.rawValueSize && numDataBlocks == that.numDataBlocks
&& numEntries == that.numEntries && numDeletions == that.numDeletions
&& numMergeOperands == that.numMergeOperands && numRangeDeletions == that.numRangeDeletions
&& formatVersion == that.formatVersion && fixedKeyLen == that.fixedKeyLen
&& columnFamilyId == that.columnFamilyId && creationTime == that.creationTime
&& oldestKeyTime == that.oldestKeyTime
&& Arrays.equals(columnFamilyName, that.columnFamilyName)
&& Objects.equals(filterPolicyName, that.filterPolicyName)
&& Objects.equals(comparatorName, that.comparatorName)
&& Objects.equals(mergeOperatorName, that.mergeOperatorName)
&& Objects.equals(prefixExtractorName, that.prefixExtractorName)
&& Objects.equals(propertyCollectorsNames, that.propertyCollectorsNames)
&& Objects.equals(compressionName, that.compressionName)
&& Objects.equals(userCollectedProperties, that.userCollectedProperties)
&& Objects.equals(readableProperties, that.readableProperties)
&& Objects.equals(propertiesOffsets, that.propertiesOffsets);
}
@Override
public int hashCode() {
int result = Objects.hash(dataSize, indexSize, indexPartitions, topLevelIndexSize,
indexKeyIsUserKey, indexValueIsDeltaEncoded, filterSize, rawKeySize, rawValueSize,
numDataBlocks, numEntries, numDeletions, numMergeOperands, numRangeDeletions, formatVersion,
fixedKeyLen, columnFamilyId, creationTime, oldestKeyTime, filterPolicyName, comparatorName,
mergeOperatorName, prefixExtractorName, propertyCollectorsNames, compressionName,
userCollectedProperties, readableProperties, propertiesOffsets);
result = 31 * result + Arrays.hashCode(columnFamilyName);
return result;
}
}

44
java/src/main/java/org/rocksdb/WriteStallCondition.java Normal file
View File

@ -0,0 +1,44 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
public enum WriteStallCondition {
NORMAL((byte) 0x0),
DELAYED((byte) 0x1),
STOPPED((byte) 0x2);
private final byte value;
WriteStallCondition(final byte value) {
this.value = value;
}
/**
* Get the internal representation.
*
* @return the internal representation
*/
byte getValue() {
return value;
}
/**
* Get the WriteStallCondition from the internal representation value.
*
* @return the flush reason.
*
* @throws IllegalArgumentException if the value is unknown.
*/
static WriteStallCondition fromValue(final byte value) {
for (final WriteStallCondition writeStallCondition : WriteStallCondition.values()) {
if (writeStallCondition.value == value) {
return writeStallCondition;
}
}
throw new IllegalArgumentException("Illegal value provided for WriteStallCondition: " + value);
}
}

75
java/src/main/java/org/rocksdb/WriteStallInfo.java Normal file
View File

@ -0,0 +1,75 @@
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
package org.rocksdb;
import java.util.Objects;
public class WriteStallInfo {
private final String columnFamilyName;
private final WriteStallCondition currentCondition;
private final WriteStallCondition previousCondition;
/**
* Access is package private as this will only be constructed from
* C++ via JNI and for testing.
*/
WriteStallInfo(final String columnFamilyName, final byte currentConditionValue,
final byte previousConditionValue) {
this.columnFamilyName = columnFamilyName;
this.currentCondition = WriteStallCondition.fromValue(currentConditionValue);
this.previousCondition = WriteStallCondition.fromValue(previousConditionValue);
}
/**
* Get the name of the column family.
*
* @return the name of the column family.
*/
public String getColumnFamilyName() {
return columnFamilyName;
}
/**
* Get the current state of the write controller.
*
* @return the current state.
*/
public WriteStallCondition getCurrentCondition() {
return currentCondition;
}
/**
* Get the previous state of the write controller.
*
* @return the previous state.
*/
public WriteStallCondition getPreviousCondition() {
return previousCondition;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
WriteStallInfo that = (WriteStallInfo) o;
return Objects.equals(columnFamilyName, that.columnFamilyName)
&& currentCondition == that.currentCondition && previousCondition == that.previousCondition;
}
@Override
public int hashCode() {
return Objects.hash(columnFamilyName, currentCondition, previousCondition);
}
@Override
public String toString() {
return "WriteStallInfo{"
+ "columnFamilyName='" + columnFamilyName + '\'' + ", currentCondition=" + currentCondition
+ ", previousCondition=" + previousCondition + '}';
}
}

45
java/src/test/java/org/rocksdb/DBOptionsTest.java
View File

@ -5,13 +5,16 @@
package org.rocksdb;
import org.junit.ClassRule;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.nio.file.Paths;
import java.util.*;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.ClassRule;
import org.junit.Test;
public class DBOptionsTest {
@ -895,4 +898,38 @@ public class DBOptionsTest {
assertThat(options.skipCheckingSstFileSizesOnDbOpen()).isEqualTo(true);
}
}
@Test
public void eventListeners() {
final AtomicBoolean wasCalled1 = new AtomicBoolean();
final AtomicBoolean wasCalled2 = new AtomicBoolean();
try (final DBOptions options = new DBOptions();
final AbstractEventListener el1 =
new AbstractEventListener() {
@Override
public void onTableFileDeleted(final TableFileDeletionInfo tableFileDeletionInfo) {
wasCalled1.set(true);
}
};
final AbstractEventListener el2 =
new AbstractEventListener() {
@Override
public void onMemTableSealed(final MemTableInfo memTableInfo) {
wasCalled2.set(true);
}
}) {
assertThat(options.setListeners(Arrays.asList(el1, el2))).isEqualTo(options);
List<AbstractEventListener> listeners = options.listeners();
assertEquals(el1, listeners.get(0));
assertEquals(el2, listeners.get(1));
options.setListeners(Collections.<AbstractEventListener>emptyList());
listeners.get(0).onTableFileDeleted(null);
assertTrue(wasCalled1.get());
listeners.get(1).onMemTableSealed(null);
assertTrue(wasCalled2.get());
List<AbstractEventListener> listeners2 = options.listeners();
assertNotNull(listeners2);
assertEquals(0, listeners2.size());
}
}
}

589
java/src/test/java/org/rocksdb/EventListenerTest.java Normal file
View File

@ -0,0 +1,589 @@
package org.rocksdb;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.*;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collections;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.rocksdb.test.TestableEventListener;
public class EventListenerTest {
@ClassRule
public static final RocksNativeLibraryResource ROCKS_NATIVE_LIBRARY_RESOURCE =
new RocksNativeLibraryResource();
@Rule public TemporaryFolder dbFolder = new TemporaryFolder();
public static final Random rand = PlatformRandomHelper.getPlatformSpecificRandomFactory();
void flushDb(final AbstractEventListener el, final AtomicBoolean wasCbCalled)
throws RocksDBException {
try (final Options opt =
new Options().setCreateIfMissing(true).setListeners(Collections.singletonList(el));
final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
assertThat(db).isNotNull();
final byte[] value = new byte[24];
rand.nextBytes(value);
db.put("testKey".getBytes(), value);
db.flush(new FlushOptions());
assertTrue(wasCbCalled.get());
}
}
@Test
public void onFlushCompleted() throws RocksDBException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onFlushCompletedListener = new AbstractEventListener() {
@Override
public void onFlushCompleted(final RocksDB rocksDb, final FlushJobInfo flushJobInfo) {
assertNotNull(flushJobInfo.getColumnFamilyName());
assertEquals(FlushReason.MANUAL_FLUSH, flushJobInfo.getFlushReason());
wasCbCalled.set(true);
}
};
flushDb(onFlushCompletedListener, wasCbCalled);
}
@Test
public void onFlushBegin() throws RocksDBException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onFlushBeginListener = new AbstractEventListener() {
@Override
public void onFlushBegin(final RocksDB rocksDb, final FlushJobInfo flushJobInfo) {
assertNotNull(flushJobInfo.getColumnFamilyName());
assertEquals(FlushReason.MANUAL_FLUSH, flushJobInfo.getFlushReason());
wasCbCalled.set(true);
}
};
flushDb(onFlushBeginListener, wasCbCalled);
}
void deleteTableFile(final AbstractEventListener el, final AtomicBoolean wasCbCalled)
throws RocksDBException, InterruptedException {
try (final Options opt =
new Options().setCreateIfMissing(true).setListeners(Collections.singletonList(el));
final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
assertThat(db).isNotNull();
final byte[] value = new byte[24];
rand.nextBytes(value);
db.put("testKey".getBytes(), value);
RocksDB.LiveFiles liveFiles = db.getLiveFiles();
assertNotNull(liveFiles);
assertNotNull(liveFiles.files);
assertFalse(liveFiles.files.isEmpty());
db.deleteFile(liveFiles.files.get(0));
assertTrue(wasCbCalled.get());
}
}
@Test
public void onTableFileDeleted() throws RocksDBException, InterruptedException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onTableFileDeletedListener = new AbstractEventListener() {
@Override
public void onTableFileDeleted(final TableFileDeletionInfo tableFileDeletionInfo) {
assertNotNull(tableFileDeletionInfo.getDbName());
wasCbCalled.set(true);
}
};
deleteTableFile(onTableFileDeletedListener, wasCbCalled);
}
void compactRange(final AbstractEventListener el, final AtomicBoolean wasCbCalled)
throws RocksDBException {
try (final Options opt =
new Options().setCreateIfMissing(true).setListeners(Collections.singletonList(el));
final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
assertThat(db).isNotNull();
final byte[] value = new byte[24];
rand.nextBytes(value);
db.put("testKey".getBytes(), value);
db.compactRange();
assertTrue(wasCbCalled.get());
}
}
@Test
public void onCompactionBegin() throws RocksDBException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onCompactionBeginListener = new AbstractEventListener() {
@Override
public void onCompactionBegin(final RocksDB db, final CompactionJobInfo compactionJobInfo) {
assertEquals(CompactionReason.kManualCompaction, compactionJobInfo.compactionReason());
wasCbCalled.set(true);
}
};
compactRange(onCompactionBeginListener, wasCbCalled);
}
@Test
public void onCompactionCompleted() throws RocksDBException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onCompactionCompletedListener = new AbstractEventListener() {
@Override
public void onCompactionCompleted(
final RocksDB db, final CompactionJobInfo compactionJobInfo) {
assertEquals(CompactionReason.kManualCompaction, compactionJobInfo.compactionReason());
wasCbCalled.set(true);
}
};
compactRange(onCompactionCompletedListener, wasCbCalled);
}
@Test
public void onTableFileCreated() throws RocksDBException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onTableFileCreatedListener = new AbstractEventListener() {
@Override
public void onTableFileCreated(final TableFileCreationInfo tableFileCreationInfo) {
assertEquals(TableFileCreationReason.FLUSH, tableFileCreationInfo.getReason());
wasCbCalled.set(true);
}
};
flushDb(onTableFileCreatedListener, wasCbCalled);
}
@Test
public void onTableFileCreationStarted() throws RocksDBException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onTableFileCreationStartedListener = new AbstractEventListener() {
@Override
public void onTableFileCreationStarted(
final TableFileCreationBriefInfo tableFileCreationBriefInfo) {
assertEquals(TableFileCreationReason.FLUSH, tableFileCreationBriefInfo.getReason());
wasCbCalled.set(true);
}
};
flushDb(onTableFileCreationStartedListener, wasCbCalled);
}
void deleteColumnFamilyHandle(final AbstractEventListener el, final AtomicBoolean wasCbCalled)
throws RocksDBException {
try (final Options opt =
new Options().setCreateIfMissing(true).setListeners(Collections.singletonList(el));
final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
assertThat(db).isNotNull();
final byte[] value = new byte[24];
rand.nextBytes(value);
db.put("testKey".getBytes(), value);
ColumnFamilyHandle columnFamilyHandle = db.getDefaultColumnFamily();
columnFamilyHandle.close();
assertTrue(wasCbCalled.get());
}
}
@Test
public void onColumnFamilyHandleDeletionStarted() throws RocksDBException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onColumnFamilyHandleDeletionStartedListener =
new AbstractEventListener() {
@Override
public void onColumnFamilyHandleDeletionStarted(
final ColumnFamilyHandle columnFamilyHandle) {
assertNotNull(columnFamilyHandle);
wasCbCalled.set(true);
}
};
deleteColumnFamilyHandle(onColumnFamilyHandleDeletionStartedListener, wasCbCalled);
}
void ingestExternalFile(final AbstractEventListener el, final AtomicBoolean wasCbCalled)
throws RocksDBException {
try (final Options opt =
new Options().setCreateIfMissing(true).setListeners(Collections.singletonList(el));
final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
assertThat(db).isNotNull();
String uuid = UUID.randomUUID().toString();
SstFileWriter sstFileWriter = new SstFileWriter(new EnvOptions(), opt);
Path externalFilePath = Paths.get(db.getName(), uuid);
sstFileWriter.open(externalFilePath.toString());
sstFileWriter.put("testKey".getBytes(), uuid.getBytes());
sstFileWriter.finish();
db.ingestExternalFile(
Collections.singletonList(externalFilePath.toString()), new IngestExternalFileOptions());
assertTrue(wasCbCalled.get());
}
}
@Test
public void onExternalFileIngested() throws RocksDBException {
// Callback is synchronous, but we need mutable container to update boolean value in other
// method
final AtomicBoolean wasCbCalled = new AtomicBoolean();
AbstractEventListener onExternalFileIngestedListener = new AbstractEventListener() {
@Override
public void onExternalFileIngested(
final RocksDB db, final ExternalFileIngestionInfo externalFileIngestionInfo) {
assertNotNull(db);
wasCbCalled.set(true);
}
};
ingestExternalFile(onExternalFileIngestedListener, wasCbCalled);
}
@Test
public void testAllCallbacksInvocation() {
final int TEST_INT_VAL = Integer.MAX_VALUE;
final long TEST_LONG_VAL = Long.MAX_VALUE;
// Expected test data objects
final Map<String, String> userCollectedPropertiesTestData =
Collections.singletonMap("key", "value");
final Map<String, String> readablePropertiesTestData = Collections.singletonMap("key", "value");
final Map<String, Long> propertiesOffsetsTestData =
Collections.singletonMap("key", TEST_LONG_VAL);
final TableProperties tablePropertiesTestData = new TableProperties(TEST_LONG_VAL,
TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL,
TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL,
TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL,
"columnFamilyName".getBytes(), "filterPolicyName", "comparatorName", "mergeOperatorName",
"prefixExtractorName", "propertyCollectorsNames", "compressionName",
userCollectedPropertiesTestData, readablePropertiesTestData, propertiesOffsetsTestData);
final FlushJobInfo flushJobInfoTestData = new FlushJobInfo(TEST_INT_VAL, "testColumnFamily",
"/file/path", TEST_LONG_VAL, TEST_INT_VAL, true, true, TEST_LONG_VAL, TEST_LONG_VAL,
tablePropertiesTestData, (byte) 0x0a);
final Status statusTestData = new Status(Status.Code.Incomplete, Status.SubCode.NoSpace, null);
final TableFileDeletionInfo tableFileDeletionInfoTestData =
new TableFileDeletionInfo("dbName", "/file/path", TEST_INT_VAL, statusTestData);
final TableFileCreationInfo tableFileCreationInfoTestData =
new TableFileCreationInfo(TEST_LONG_VAL, tablePropertiesTestData, statusTestData, "dbName",
"columnFamilyName", "/file/path", TEST_INT_VAL, (byte) 0x03);
final TableFileCreationBriefInfo tableFileCreationBriefInfoTestData =
new TableFileCreationBriefInfo(
"dbName", "columnFamilyName", "/file/path", TEST_INT_VAL, (byte) 0x03);
final MemTableInfo memTableInfoTestData = new MemTableInfo(
"columnFamilyName", TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL, TEST_LONG_VAL);
final FileOperationInfo fileOperationInfoTestData = new FileOperationInfo("/file/path",
TEST_LONG_VAL, TEST_LONG_VAL, 1_600_699_420_000_000_000L, 5_000_000_000L, statusTestData);
final WriteStallInfo writeStallInfoTestData =
new WriteStallInfo("columnFamilyName", (byte) 0x1, (byte) 0x2);
final ExternalFileIngestionInfo externalFileIngestionInfoTestData =
new ExternalFileIngestionInfo("columnFamilyName", "/external/file/path",
"/internal/file/path", TEST_LONG_VAL, tablePropertiesTestData);
final int CALLBACKS_COUNT = 22;
final AtomicBoolean[] wasCalled = new AtomicBoolean[CALLBACKS_COUNT];
for (int i = 0; i < CALLBACKS_COUNT; ++i) {
wasCalled[i] = new AtomicBoolean();
}
TestableEventListener listener = new TestableEventListener() {
@Override
public void onFlushCompleted(final RocksDB db, final FlushJobInfo flushJobInfo) {
assertEquals(flushJobInfoTestData, flushJobInfo);
wasCalled[0].set(true);
}
@Override
public void onFlushBegin(final RocksDB db, final FlushJobInfo flushJobInfo) {
assertEquals(flushJobInfoTestData, flushJobInfo);
wasCalled[1].set(true);
}
@Override
public void onTableFileDeleted(final TableFileDeletionInfo tableFileDeletionInfo) {
assertEquals(tableFileDeletionInfoTestData, tableFileDeletionInfo);
wasCalled[2].set(true);
}
@Override
public void onCompactionBegin(final RocksDB db, final CompactionJobInfo compactionJobInfo) {
assertArrayEquals(
"compactionColumnFamily".getBytes(), compactionJobInfo.columnFamilyName());
assertEquals(statusTestData, compactionJobInfo.status());
assertEquals(TEST_LONG_VAL, compactionJobInfo.threadId());
assertEquals(TEST_INT_VAL, compactionJobInfo.jobId());
assertEquals(TEST_INT_VAL, compactionJobInfo.baseInputLevel());
assertEquals(TEST_INT_VAL, compactionJobInfo.outputLevel());
assertEquals(Collections.singletonList("inputFile.sst"), compactionJobInfo.inputFiles());
assertEquals(Collections.singletonList("outputFile.sst"), compactionJobInfo.outputFiles());
assertEquals(Collections.singletonMap("tableProperties", tablePropertiesTestData),
compactionJobInfo.tableProperties());
assertEquals(CompactionReason.kFlush, compactionJobInfo.compactionReason());
assertEquals(CompressionType.SNAPPY_COMPRESSION, compactionJobInfo.compression());
wasCalled[3].set(true);
}
@Override
public void onCompactionCompleted(
final RocksDB db, final CompactionJobInfo compactionJobInfo) {
assertArrayEquals(
"compactionColumnFamily".getBytes(), compactionJobInfo.columnFamilyName());
assertEquals(statusTestData, compactionJobInfo.status());
assertEquals(TEST_LONG_VAL, compactionJobInfo.threadId());
assertEquals(TEST_INT_VAL, compactionJobInfo.jobId());
assertEquals(TEST_INT_VAL, compactionJobInfo.baseInputLevel());
assertEquals(TEST_INT_VAL, compactionJobInfo.outputLevel());
assertEquals(Collections.singletonList("inputFile.sst"), compactionJobInfo.inputFiles());
assertEquals(Collections.singletonList("outputFile.sst"), compactionJobInfo.outputFiles());
assertEquals(Collections.singletonMap("tableProperties", tablePropertiesTestData),
compactionJobInfo.tableProperties());
assertEquals(CompactionReason.kFlush, compactionJobInfo.compactionReason());
assertEquals(CompressionType.SNAPPY_COMPRESSION, compactionJobInfo.compression());
wasCalled[4].set(true);
}
@Override
public void onTableFileCreated(final TableFileCreationInfo tableFileCreationInfo) {
assertEquals(tableFileCreationInfoTestData, tableFileCreationInfo);
wasCalled[5].set(true);
}
@Override
public void onTableFileCreationStarted(
final TableFileCreationBriefInfo tableFileCreationBriefInfo) {
assertEquals(tableFileCreationBriefInfoTestData, tableFileCreationBriefInfo);
wasCalled[6].set(true);
}
@Override
public void onMemTableSealed(final MemTableInfo memTableInfo) {
assertEquals(memTableInfoTestData, memTableInfo);
wasCalled[7].set(true);
}
@Override
public void onColumnFamilyHandleDeletionStarted(final ColumnFamilyHandle columnFamilyHandle) {
wasCalled[8].set(true);
}
@Override
public void onExternalFileIngested(
final RocksDB db, final ExternalFileIngestionInfo externalFileIngestionInfo) {
assertEquals(externalFileIngestionInfoTestData, externalFileIngestionInfo);
wasCalled[9].set(true);
}
@Override
public void onBackgroundError(
final BackgroundErrorReason backgroundErrorReason, final Status backgroundError) {
wasCalled[10].set(true);
}
@Override
public void onStallConditionsChanged(final WriteStallInfo writeStallInfo) {
assertEquals(writeStallInfoTestData, writeStallInfo);
wasCalled[11].set(true);
}
@Override
public void onFileReadFinish(final FileOperationInfo fileOperationInfo) {
assertEquals(fileOperationInfoTestData, fileOperationInfo);
wasCalled[12].set(true);
}
@Override
public void onFileWriteFinish(final FileOperationInfo fileOperationInfo) {
assertEquals(fileOperationInfoTestData, fileOperationInfo);
wasCalled[13].set(true);
}
@Override
public void OnFileFlushFinish(final FileOperationInfo fileOperationInfo) {
assertEquals(fileOperationInfoTestData, fileOperationInfo);
wasCalled[14].set(true);
}
@Override
public void OnFileSyncFinish(final FileOperationInfo fileOperationInfo) {
assertEquals(fileOperationInfoTestData, fileOperationInfo);
wasCalled[15].set(true);
}
@Override
public void OnFileRangeSyncFinish(final FileOperationInfo fileOperationInfo) {
assertEquals(fileOperationInfoTestData, fileOperationInfo);
wasCalled[16].set(true);
}
@Override
public void OnFileTruncateFinish(final FileOperationInfo fileOperationInfo) {
assertEquals(fileOperationInfoTestData, fileOperationInfo);
wasCalled[17].set(true);
}
@Override
public void OnFileCloseFinish(final FileOperationInfo fileOperationInfo) {
assertEquals(fileOperationInfoTestData, fileOperationInfo);
wasCalled[18].set(true);
}
@Override
public boolean shouldBeNotifiedOnFileIO() {
wasCalled[19].set(true);
return false;
}
@Override
public boolean onErrorRecoveryBegin(
final BackgroundErrorReason backgroundErrorReason, final Status backgroundError) {
assertEquals(BackgroundErrorReason.FLUSH, backgroundErrorReason);
assertEquals(statusTestData, backgroundError);
wasCalled[20].set(true);
return true;
}
@Override
public void onErrorRecoveryCompleted(final Status oldBackgroundError) {
assertEquals(statusTestData, oldBackgroundError);
wasCalled[21].set(true);
}
};
listener.invokeAllCallbacks();
for (int i = 0; i < CALLBACKS_COUNT; ++i) {
assertTrue("Callback method " + i + " was not called", wasCalled[i].get());
}
}
@Test
public void testEnabledCallbacks() {
final AtomicBoolean wasOnMemTableSealedCalled = new AtomicBoolean();
final AtomicBoolean wasOnErrorRecoveryCompletedCalled = new AtomicBoolean();
final TestableEventListener listener = new TestableEventListener(
AbstractEventListener.EnabledEventCallback.ON_MEMTABLE_SEALED,
AbstractEventListener.EnabledEventCallback.ON_ERROR_RECOVERY_COMPLETED) {
@Override
public void onFlushCompleted(final RocksDB db, final FlushJobInfo flushJobInfo) {
fail("onFlushCompleted was not enabled");
}
@Override
public void onFlushBegin(final RocksDB db, final FlushJobInfo flushJobInfo) {
fail("onFlushBegin was not enabled");
}
@Override
public void onTableFileDeleted(final TableFileDeletionInfo tableFileDeletionInfo) {
fail("onTableFileDeleted was not enabled");
}
@Override
public void onCompactionBegin(final RocksDB db, final CompactionJobInfo compactionJobInfo) {
fail("onCompactionBegin was not enabled");
}
@Override
public void onCompactionCompleted(
final RocksDB db, final CompactionJobInfo compactionJobInfo) {
fail("onCompactionCompleted was not enabled");
}
@Override
public void onTableFileCreated(final TableFileCreationInfo tableFileCreationInfo) {
fail("onTableFileCreated was not enabled");
}
@Override
public void onTableFileCreationStarted(
final TableFileCreationBriefInfo tableFileCreationBriefInfo) {
fail("onTableFileCreationStarted was not enabled");
}
@Override
public void onMemTableSealed(final MemTableInfo memTableInfo) {
wasOnMemTableSealedCalled.set(true);
}
@Override
public void onColumnFamilyHandleDeletionStarted(final ColumnFamilyHandle columnFamilyHandle) {
fail("onColumnFamilyHandleDeletionStarted was not enabled");
}
@Override
public void onExternalFileIngested(
final RocksDB db, final ExternalFileIngestionInfo externalFileIngestionInfo) {
fail("onExternalFileIngested was not enabled");
}
@Override
public void onBackgroundError(
final BackgroundErrorReason backgroundErrorReason, final Status backgroundError) {
fail("onBackgroundError was not enabled");
}
@Override
public void onStallConditionsChanged(final WriteStallInfo writeStallInfo) {
fail("onStallConditionsChanged was not enabled");
}
@Override
public void onFileReadFinish(final FileOperationInfo fileOperationInfo) {
fail("onFileReadFinish was not enabled");
}
@Override
public void onFileWriteFinish(final FileOperationInfo fileOperationInfo) {
fail("onFileWriteFinish was not enabled");
}
@Override
public void OnFileFlushFinish(final FileOperationInfo fileOperationInfo) {
fail("OnFileFlushFinish was not enabled");
}
@Override
public void OnFileSyncFinish(final FileOperationInfo fileOperationInfo) {
fail("OnFileSyncFinish was not enabled");
}
@Override
public void OnFileRangeSyncFinish(final FileOperationInfo fileOperationInfo) {
fail("OnFileRangeSyncFinish was not enabled");
}
@Override
public void OnFileTruncateFinish(final FileOperationInfo fileOperationInfo) {
fail("OnFileTruncateFinish was not enabled");
}
@Override
public void OnFileCloseFinish(final FileOperationInfo fileOperationInfo) {
fail("OnFileCloseFinish was not enabled");
}
@Override
public boolean shouldBeNotifiedOnFileIO() {
fail("shouldBeNotifiedOnFileIO was not enabled");
return false;
}
@Override
public boolean onErrorRecoveryBegin(
final BackgroundErrorReason backgroundErrorReason, final Status backgroundError) {
fail("onErrorRecoveryBegin was not enabled");
return true;
}
@Override
public void onErrorRecoveryCompleted(final Status oldBackgroundError) {
wasOnErrorRecoveryCompletedCalled.set(true);
}
};
listener.invokeAllCallbacks();
assertTrue(wasOnMemTableSealedCalled.get());
assertTrue(wasOnErrorRecoveryCompletedCalled.get());
}
}

38
java/src/test/java/org/rocksdb/OptionsTest.java
View File

@ -6,13 +6,13 @@
package org.rocksdb;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.*;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.ClassRule;
import org.junit.Test;
import org.rocksdb.test.RemoveEmptyValueCompactionFilterFactory;
@ -1436,4 +1436,38 @@ public class OptionsTest {
assertThat(options.skipCheckingSstFileSizesOnDbOpen()).isEqualTo(true);
}
}
@Test
public void eventListeners() {
final AtomicBoolean wasCalled1 = new AtomicBoolean();
final AtomicBoolean wasCalled2 = new AtomicBoolean();
try (final Options options = new Options();
final AbstractEventListener el1 =
new AbstractEventListener() {
@Override
public void onTableFileDeleted(final TableFileDeletionInfo tableFileDeletionInfo) {
wasCalled1.set(true);
}
};
final AbstractEventListener el2 =
new AbstractEventListener() {
@Override
public void onMemTableSealed(final MemTableInfo memTableInfo) {
wasCalled2.set(true);
}
}) {
assertThat(options.setListeners(Arrays.asList(el1, el2))).isEqualTo(options);
List<AbstractEventListener> listeners = options.listeners();
assertEquals(el1, listeners.get(0));
assertEquals(el2, listeners.get(1));
options.setListeners(Collections.<AbstractEventListener>emptyList());
listeners.get(0).onTableFileDeleted(null);
assertTrue(wasCalled1.get());
listeners.get(1).onMemTableSealed(null);
assertTrue(wasCalled2.get());
List<AbstractEventListener> listeners2 = options.listeners();
assertNotNull(listeners2);
assertEquals(0, listeners2.size());
}
}
}

19
java/src/test/java/org/rocksdb/test/TestableEventListener.java Normal file
View File

@ -0,0 +1,19 @@
package org.rocksdb.test;
import org.rocksdb.AbstractEventListener;
public class TestableEventListener extends AbstractEventListener {
public TestableEventListener() {
super();
}
public TestableEventListener(final EnabledEventCallback... enabledEventCallbacks) {
super(enabledEventCallbacks);
}
public void invokeAllCallbacks() {
invokeAllCallbacks(nativeHandle_);
}
private static native void invokeAllCallbacks(final long handle);
}

3
src.mk
View File

@ -552,6 +552,8 @@ JNI_NATIVE_SOURCES = \
java/rocksjni/config_options.cc \
java/rocksjni/env.cc \
java/rocksjni/env_options.cc \
java/rocksjni/event_listener.cc \
java/rocksjni/event_listener_jnicallback.cc \
java/rocksjni/ingest_external_file_options.cc \
java/rocksjni/filter.cc \
java/rocksjni/iterator.cc \
@ -598,6 +600,7 @@ JNI_NATIVE_SOURCES = \
java/rocksjni/transaction_notifier.cc \
java/rocksjni/transaction_notifier_jnicallback.cc \
java/rocksjni/ttl.cc \
java/rocksjni/testable_event_listener.cc \
java/rocksjni/wal_filter.cc \
java/rocksjni/wal_filter_jnicallback.cc \
java/rocksjni/write_batch.cc \