ESPushTask extends BaseTask { protected OutgoingMessageEnvelope getAvroKeyOutMsg(IncomingMessageEnvelope envelope, ESPushTaskConfig.ESIndexSpec spec) { return getAvroKeyOutMsg(envelope, spec, Optional.empty()); }  static void main(String [] args);  }
@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()); }
@Test public void testDocMissing() throws Exception { ESPushTaskConfig.ESIndexSpec esConfig = getEsIndexSpec("key_avro", false); ESPushTask task = getEsPushTask(); ActionRequestKey inKeyUpdate = ActionRequestKey.newBuilder() .setAction(Action.UPDATE) .setId("foo") .build(); when(task.avroSerde.fromBytes(null)).thenReturn(inKeyUpdate); assertThatThrownBy(() -> task.getAvroKeyOutMsg(getInMsg(null), esConfig)).isInstanceOf(InvalidParameterException.class) .hasMessageContaining("Document must be provided"); ActionRequestKey inKeyIndex = ActionRequestKey.newBuilder() .setAction(Action.INDEX) .setId("foo") .build(); when(task.avroSerde.fromBytes(null)).thenReturn(inKeyIndex); assertThatThrownBy(() -> task.getAvroKeyOutMsg(getInMsg(null), esConfig)).isInstanceOf(InvalidParameterException.class) .hasMessageContaining("Document must be provided"); }
@Test public void testWritesWithToStaticIndex() throws Exception { ESPushTaskConfig.ESIndexSpec esConfig = getEsIndexSpec("key_avro", false); ESPushTask task = getEsPushTask(); ActionRequestKey inKeyMissingId = ActionRequestKey.newBuilder() .setAction(Action.DELETE) .build(); when(task.avroSerde.fromBytes(null)).thenReturn(inKeyMissingId); assertThatThrownBy(() -> task.getAvroKeyOutMsg(getInMsg(""), esConfig)).isInstanceOf(InvalidParameterException.class) .hasMessageContaining("Document id is required"); ActionRequestKey inKey = ActionRequestKey.newBuilder() .setAction(Action.UPDATE) .setId("blah") .build(); when(task.avroSerde.fromBytes(null)).thenReturn(inKey); OutgoingMessageEnvelope out = task.getAvroKeyOutMsg(getInMsg("{}"), esConfig); HTTPBulkLoader.ActionRequest req = (HTTPBulkLoader.ActionRequest) out.getMessage(); assertEquals("blah", req.key.getId().toString()); assertEquals(Action.UPDATE, req.key.getAction()); assertNull("Do not default partition time", req.key.getPartitionTsUnixMs()); assertNull("Do not default event time", req.key.getEventTsUnixMs()); assertNull("No version set", req.key.getVersion()); assertNull("No version type set", req.key.getVersionType()); assertEquals("{\"doc\":{}}", req.document); }
@Test public void testDefaultDocIdWithAvroKeyConfig() throws Exception { ESPushTaskConfig.ESIndexSpec esConfig = getEsIndexSpec("key_avro", true); ESPushTask task = getEsPushTask(); ActionRequestKey inKey = ActionRequestKey.newBuilder() .setEventTsUnixMs(3L) .setPartitionTsUnixMs(4L) .setVersionType(VersionType.EXTERNAL) .setVersion(5L) .build(); when(task.avroSerde.fromBytes(null)).thenReturn(inKey); OutgoingMessageEnvelope out = task.getAvroKeyOutMsg(getInMsg(""), esConfig); HTTPBulkLoader.ActionRequest req = (HTTPBulkLoader.ActionRequest) out.getMessage(); assertEquals("fake-0-1234", req.key.getId().toString()); assertEquals(Action.INDEX, req.key.getAction()); assertEquals(4L, req.key.getPartitionTsUnixMs().longValue()); assertEquals(3L, req.key.getEventTsUnixMs().longValue()); assertEquals(VersionType.EXTERNAL, req.key.getVersionType()); assertEquals(5L, req.key.getVersion().longValue()); }