// an error immediately to the caller
		atomic.AddInt64(&h.totalMessages, 1)
		atomic.AddInt64(&h.failedMessages, 1)
		return errors.New("log buffer full")
	}
	return nil
}

// SendFromStore - reads the log from store and sends it to kafka.
func (h *Target) SendFromStore(key store.Key) (err error) {
	auditEntry, err := h.store.Get(key)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}

     */
    public void writeOctetArray(final byte[] b, final int i, final int l) {
        System.arraycopy(b, i, this.buf, this.index, l);
        advance(l);
    }

    /**
     * Reads an octet array from the buffer.
     *
     * @param b the byte array to read into
     * @param i the starting index in the array
     * @param l the number of bytes to read
     */

        System.out.println("RDMA Statistics: " + stats);

        // Test statistics reset functionality
        stats.reset();
        assertEquals(0, stats.getRdmaReads(), "Reads should be reset to 0");
        assertEquals(0, stats.getRdmaWrites(), "Writes should be reset to 0");
        assertEquals(0, stats.getBytesTransferred(), "Bytes should be reset to 0");

		if err != nil {
			return err
		}

		_, err = target.insertStmt.Exec(eventTime, data)

		return err
	}

	return nil
}

// SendFromStore - reads an event from store and sends it to MySQL.
func (target *MySQLTarget) SendFromStore(key store.Key) error {
	if err := target.init(); err != nil {
		return err
	}

	_, err := target.isActive()
	if err != nil {

            shift += 7
          } else {
            result += b shl shift // Last byte.
            break
          }
        }
        return result
      }

      /** Reads a potentially Huffman encoded byte string. */
      @Throws(IOException::class)
      fun readByteString(): ByteString {
        val firstByte = readByte()
        val huffmanDecode = firstByte and 0x80 == 0x80 // 1NNNNNNN

  Lockable {
  // Internal state of this connection is guarded by 'lock'. No blocking operations may be
  // performed while holding this lock!
  //
  // Socket writes are guarded by frameWriter.
  //
  // Socket reads are unguarded but are only made by the reader thread.
  //
  // Certain operations (like SYN_STREAM) need to synchronize on both the frameWriter (to do

 *
 * <p>{@link #setDefault} allows subclasses to specify default values for types.
 *
 * <p>This class incurs IO because it scans the classpath and reads classpath resources.
 *
 * @author Ben Yu
 * @since 14.0
 */
// TODO: Switch to JUnit 4 and use @Parameterized and @BeforeClass
// Note: @Test annotations are deliberate, as some subclasses specify @RunWith(JUnit4).

        }
    }

    @Nested
    @DisplayName("Integration Tests")
    class IntegrationTests {

        @Test
        @DisplayName("Should handle multiple consecutive reads")
        void testMultipleConsecutiveReads() throws Exception {
            // First read
            byte[] buffer1 = createValidOplockBreakBuffer((byte) 0x01, createTestData(16));

                break;
            totalRead += read;
        }

        assertEquals(size, totalRead);
        assertArrayEquals(data, buffer);
    }

    @Test
    @DisplayName("Read handles partial reads from underlying stream")
    void readHandlesPartialReads() throws IOException {
        // Create message with data
        byte[] data = new byte[] { 1, 2, 3, 4, 5 };

			_, err = target.jstream.Publish(target.args.Subject, data)
		} else {
			err = target.natsConn.Publish(target.args.Subject, data)
		}
	}
	return err
}

// SendFromStore - reads an event from store and sends it to Nats.
func (target *NATSTarget) SendFromStore(key store.Key) error {
	if err := target.init(); err != nil {
		return err
	}

	_, err := target.isActive()
	if err != nil {