diff --git a/java/rocksjni/slice.cc b/java/rocksjni/slice.cc
index 583d745661c2..1d22f1eb6ca8 100644
--- a/java/rocksjni/slice.cc
+++ b/java/rocksjni/slice.cc
@@ -267,9 +267,11 @@ void Java_org_rocksdb_Slice_disposeInternalBuf(JNIEnv* /*env*/, jclass /*jcls*/,
 jlong Java_org_rocksdb_DirectSlice_createNewDirectSlice0(JNIEnv* env,
                                                          jclass /*jcls*/,
                                                          jobject data,
+                                                         jint offset,
                                                          jint length) {
   assert(data != nullptr);
   void* data_addr = env->GetDirectBufferAddress(data);
+  jlong capacity = env->GetDirectBufferCapacity(data);
   if (data_addr == nullptr) {
     // error: memory region is undefined, given object is not a direct
     // java.nio.Buffer, or JNI access to direct buffers is not supported by JVM
@@ -278,9 +280,17 @@ jlong Java_org_rocksdb_DirectSlice_createNewDirectSlice0(JNIEnv* env,
                  "Could not access DirectBuffer"));
     return 0;
   }
+  if ((jlong)length > capacity - offset) {
+    // error: memory region is undefined, given object is not a direct
+    // java.nio.Buffer, or JNI access to direct buffers is not supported by JVM
+    ROCKSDB_NAMESPACE::IllegalArgumentExceptionJni::ThrowNew(
+            env, ROCKSDB_NAMESPACE::Status::InvalidArgument(
+                    "Invalid Offset and Length passed"));
+    return 0;
+  }
 
   const auto* ptrData = reinterpret_cast<char*>(data_addr);
-  const auto* slice = new ROCKSDB_NAMESPACE::Slice(ptrData, length);
+  const ROCKSDB_NAMESPACE::Slice* slice = new ROCKSDB_NAMESPACE::Slice(ptrData + offset, length);
   return GET_CPLUSPLUS_POINTER(slice);
 }
 
diff --git a/java/src/main/java/org/rocksdb/DirectSlice.java b/java/src/main/java/org/rocksdb/DirectSlice.java
index 88ec29e3bd65..60efb863a80a 100644
--- a/java/src/main/java/org/rocksdb/DirectSlice.java
+++ b/java/src/main/java/org/rocksdb/DirectSlice.java
@@ -63,7 +63,19 @@ public DirectSlice(final String str) {
    * @param length The length of the data to use for the slice
    */
   public DirectSlice(final ByteBuffer data, final int length) {
-    super(createNewDirectSlice0(ensureDirect(data), length));
+    this(data, 0, length);
+  }
+
+  /**
+   * Constructs a slice where the data is
+   * read from the provided
+   * ByteBuffer up to a certain length
+   *
+   * @param data The buffer containing the data
+   * @param length The length of the data to use for the slice
+   */
+  public DirectSlice(final ByteBuffer data, final int offset, final int length) {
+    super(createNewDirectSlice0(ensureDirect(data), offset, length));
     this.internalBuffer = false;
   }
 
@@ -123,7 +135,7 @@ protected void disposeInternal() {
     disposeInternal(nativeHandle);
   }
 
-  private static native long createNewDirectSlice0(final ByteBuffer data, final int length);
+  private static native long createNewDirectSlice0(final ByteBuffer data, final int offset, final int length);
   private static native long createNewDirectSlice1(final ByteBuffer data);
   @Override protected final native ByteBuffer data0(long handle);
   private static native byte get0(long handle, int offset);
diff --git a/java/src/test/java/org/rocksdb/DirectSliceTest.java b/java/src/test/java/org/rocksdb/DirectSliceTest.java
index 67385345c520..2209e9288f76 100644
--- a/java/src/test/java/org/rocksdb/DirectSliceTest.java
+++ b/java/src/test/java/org/rocksdb/DirectSliceTest.java
@@ -4,6 +4,7 @@
 //  (found in the LICENSE.Apache file in the root directory).
 package org.rocksdb;
 
+import org.junit.Assert;
 import org.junit.ClassRule;
 import org.junit.Test;
 
@@ -54,6 +55,37 @@ public void directSliceWithByteBufferAndLength() {
     }
   }
 
+
+  @Test
+  public void directSliceWithByteBufferOffsetAndLength() {
+    final byte[] data = "Some text".getBytes();
+    final ByteBuffer buffer = ByteBuffer.allocateDirect(data.length);
+    buffer.put(data);
+    try(final DirectSlice directSlice = new DirectSlice(buffer, 2, 6)) {
+      assertThat(directSlice.toString()).isEqualTo("me tex");
+    }
+  }
+
+  @Test
+  public void directSliceWithInvalidLength() {
+    final byte[] data = "Some text".getBytes();
+    final ByteBuffer buffer = ByteBuffer.allocateDirect(data.length);
+    buffer.put(data);
+    Assert.assertThrows(IllegalArgumentException.class, () -> {
+      new DirectSlice(buffer, 2, 100);
+    });
+  }
+
+  @Test
+  public void directSliceWithInvalidOffset() {
+    final byte[] data = "Some text".getBytes();
+    final ByteBuffer buffer = ByteBuffer.allocateDirect(data.length);
+    buffer.put(data);
+    Assert.assertThrows(IllegalArgumentException.class, () -> {
+      new DirectSlice(buffer, data.length + 2, 4);
+    });
+  }
+
   @Test(expected = IllegalArgumentException.class)
   public void directSliceInitWithoutDirectAllocation() {
     final byte[] data = "Some text".getBytes();