FlinkKeyUtils { public static <K> ByteBuffer encodeKey(K key, Coder<K> keyCoder) { checkNotNull(keyCoder, "Provided coder must not be null"); final byte[] keyBytes; try { keyBytes = CoderUtils.encodeToByteArray(keyCoder, key, Coder.Context.NESTED); } catch (Exception e) { throw new RuntimeException(String.format(Locale.ENGLISH, "Failed to encode key: %s", key), e); } return ByteBuffer.wrap(keyBytes); }  static ByteBuffer encodeKey(K key, Coder<K> keyCoder); static K decodeKey(ByteBuffer byteBuffer, Coder<K> keyCoder);  }
@Test @SuppressWarnings("ByteBufferBackingArray") public void testCoderContext() throws Exception { String input = "hello world"; Coder<String> coder = StringUtf8Coder.of(); ByteBuffer encoded = FlinkKeyUtils.encodeKey(input, coder); assertThat( encoded.array(), is(CoderUtils.encodeToByteArray(coder, input, Coder.Context.NESTED))); }