@Test
public void testWritesWithPartitionedIndexes() throws Exception {

    ESPushTaskConfig.ESIndexSpec esConfig = getEsIndexSpec("key_avro", true);
    ESPushTask task = getEsPushTask();
    ActionRequestKey inKeyMissingId = ActionRequestKey.newBuilder()
        .setAction(Action.UPDATE)
        .setPartitionTsUnixMs(99L)
        .build();
    when(task.avroSerde.fromBytes(null)).thenReturn(inKeyMissingId);
    assertThatThrownBy(() -> { task.getAvroKeyOutMsg(getInMsg(""), esConfig); }).isInstanceOf(InvalidParameterException.class)
        .hasMessageContaining("Document id is required");
    ActionRequestKey inKeyMissingTs = ActionRequestKey.newBuilder()
        .setAction(Action.DELETE)
        .setId("foo")
        .build();
    when(task.avroSerde.fromBytes(null)).thenReturn(inKeyMissingTs);
    assertThatThrownBy(() -> { task.getAvroKeyOutMsg(getInMsg(""), esConfig); }).isInstanceOf(InvalidParameterException.class)
        .hasMessageContaining("Partition timestamp is required");
    ActionRequestKey inKey = ActionRequestKey.newBuilder()
        .setAction(Action.DELETE)
        .setId("foo")
        .setPartitionTsUnixMs(99L)
        .build();
    when(task.avroSerde.fromBytes(null)).thenReturn(inKey);
    OutgoingMessageEnvelope out = task.getAvroKeyOutMsg(getInMsg(""), esConfig);
    HTTPBulkLoader.ActionRequest req = (HTTPBulkLoader.ActionRequest) out.getMessage();
    assertEquals("foo", req.key.getId().toString());
    assertEquals(Action.DELETE, req.key.getAction());
    assertEquals(99L, req.key.getPartitionTsUnixMs().longValue());
    assertNull("Do not default event time", req.key.getEventTsUnixMs());
    assertNull("No version set", req.key.getVersion());
    assertNull("No version type set", req.key.getVersionType());
}