@Test public void testSendBasic() { Datastream datastream = DatastreamTestUtils.createDatastreams(DummyConnector.CONNECTOR_TYPE, "test-ds")[0]; DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); String someTopicName = "someTopicName"; AtomicInteger numEventsProduced = new AtomicInteger(); TransportProvider transport = new NoOpTransportProviderAdminFactory.NoOpTransportProvider() { @Override public void send(String destination, DatastreamProducerRecord record, SendCallback onComplete) { numEventsProduced.incrementAndGet(); DatastreamRecordMetadata metadata = new DatastreamRecordMetadata(record.getCheckpoint(), someTopicName, record.getPartition().orElse(0)); onComplete.onCompletion(metadata, null); } }; EventProducer eventProducer = new EventProducer(task, transport, new NoOpCheckpointProvider(), new Properties(), false); int eventCount = 5; for (int i = 0; i < eventCount; i++) { eventProducer.send(createDatastreamProducerRecord(), (m, e) -> { }); } Assert.assertEquals(eventCount, numEventsProduced.get()); DynamicMetricsManager metrics = DynamicMetricsManager.getInstance(); Assert.assertNotNull( metrics.getMetric("EventProducer." + someTopicName + "." + EventProducer.EVENTS_LATENCY_MS_STRING)); Assert.assertNotNull( metrics.getMetric("EventProducer." + someTopicName + "." + EventProducer.EVENTS_SEND_LATENCY_MS_STRING)); }
@Override public void send(DatastreamProducerRecord record, SendCallback sendCallback) { try { validateEventRecord(record); try { record.serializeEvents(_datastreamTask.getDestinationSerDes()); } catch (Exception e) { if (_skipMessageOnSerializationErrors) { _logger.info(String.format("Skipping the message on serialization error as configured. " + "Datastream name: %s, Datastream task name: %s", getDatastreamName(), _datastreamTask.getDatastreamTaskName()), e); _dynamicMetricsManager.createOrUpdateCounter(MODULE, getDatastreamName(), DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); _dynamicMetricsManager.createOrUpdateCounter(MODULE, AGGREGATE, DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); return; } throw e; } String destination = record.getDestination().orElse(_datastreamTask.getDatastreamDestination().getConnectionString()); record.setEventsSendTimestamp(System.currentTimeMillis()); long recordEventsSourceTimestamp = record.getEventsSourceTimestamp(); long recordEventsSendTimestamp = record.getEventsSendTimestamp().orElse(0L); _transportProvider.send(destination, record, (metadata, exception) -> onSendCallback(metadata, exception, sendCallback, recordEventsSourceTimestamp, recordEventsSendTimestamp)); } catch (Exception e) { String errorMessage = String.format("Failed send the event %s exception %s", record, e); _logger.warn(errorMessage, e); throw new DatastreamRuntimeException(errorMessage, e); } if (Instant.now().isAfter(_lastFlushTime.plus(_flushInterval))) { flush(); } }
EventProducer implements DatastreamEventProducer { @Override public void send(DatastreamProducerRecord record, SendCallback sendCallback) { try { validateEventRecord(record); try { record.serializeEvents(_datastreamTask.getDestinationSerDes()); } catch (Exception e) { if (_skipMessageOnSerializationErrors) { _logger.info(String.format("Skipping the message on serialization error as configured. " + "Datastream name: %s, Datastream task name: %s", getDatastreamName(), _datastreamTask.getDatastreamTaskName()), e); _dynamicMetricsManager.createOrUpdateCounter(MODULE, getDatastreamName(), DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); _dynamicMetricsManager.createOrUpdateCounter(MODULE, AGGREGATE, DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); return; } throw e; } String destination = record.getDestination().orElse(_datastreamTask.getDatastreamDestination().getConnectionString()); record.setEventsSendTimestamp(System.currentTimeMillis()); long recordEventsSourceTimestamp = record.getEventsSourceTimestamp(); long recordEventsSendTimestamp = record.getEventsSendTimestamp().orElse(0L); _transportProvider.send(destination, record, (metadata, exception) -> onSendCallback(metadata, exception, sendCallback, recordEventsSourceTimestamp, recordEventsSendTimestamp)); } catch (Exception e) { String errorMessage = String.format("Failed send the event %s exception %s", record, e); _logger.warn(errorMessage, e); throw new DatastreamRuntimeException(errorMessage, e); } if (Instant.now().isAfter(_lastFlushTime.plus(_flushInterval))) { flush(); } } }
EventProducer implements DatastreamEventProducer { @Override public void send(DatastreamProducerRecord record, SendCallback sendCallback) { try { validateEventRecord(record); try { record.serializeEvents(_datastreamTask.getDestinationSerDes()); } catch (Exception e) { if (_skipMessageOnSerializationErrors) { _logger.info(String.format("Skipping the message on serialization error as configured. " + "Datastream name: %s, Datastream task name: %s", getDatastreamName(), _datastreamTask.getDatastreamTaskName()), e); _dynamicMetricsManager.createOrUpdateCounter(MODULE, getDatastreamName(), DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); _dynamicMetricsManager.createOrUpdateCounter(MODULE, AGGREGATE, DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); return; } throw e; } String destination = record.getDestination().orElse(_datastreamTask.getDatastreamDestination().getConnectionString()); record.setEventsSendTimestamp(System.currentTimeMillis()); long recordEventsSourceTimestamp = record.getEventsSourceTimestamp(); long recordEventsSendTimestamp = record.getEventsSendTimestamp().orElse(0L); _transportProvider.send(destination, record, (metadata, exception) -> onSendCallback(metadata, exception, sendCallback, recordEventsSourceTimestamp, recordEventsSendTimestamp)); } catch (Exception e) { String errorMessage = String.format("Failed send the event %s exception %s", record, e); _logger.warn(errorMessage, e); throw new DatastreamRuntimeException(errorMessage, e); } if (Instant.now().isAfter(_lastFlushTime.plus(_flushInterval))) { flush(); } } EventProducer(DatastreamTask task, TransportProvider transportProvider, CheckpointProvider checkpointProvider,
      Properties config, boolean customCheckpointing); }
EventProducer implements DatastreamEventProducer { @Override public void send(DatastreamProducerRecord record, SendCallback sendCallback) { try { validateEventRecord(record); try { record.serializeEvents(_datastreamTask.getDestinationSerDes()); } catch (Exception e) { if (_skipMessageOnSerializationErrors) { _logger.info(String.format("Skipping the message on serialization error as configured. " + "Datastream name: %s, Datastream task name: %s", getDatastreamName(), _datastreamTask.getDatastreamTaskName()), e); _dynamicMetricsManager.createOrUpdateCounter(MODULE, getDatastreamName(), DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); _dynamicMetricsManager.createOrUpdateCounter(MODULE, AGGREGATE, DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); return; } throw e; } String destination = record.getDestination().orElse(_datastreamTask.getDatastreamDestination().getConnectionString()); record.setEventsSendTimestamp(System.currentTimeMillis()); long recordEventsSourceTimestamp = record.getEventsSourceTimestamp(); long recordEventsSendTimestamp = record.getEventsSendTimestamp().orElse(0L); _transportProvider.send(destination, record, (metadata, exception) -> onSendCallback(metadata, exception, sendCallback, recordEventsSourceTimestamp, recordEventsSendTimestamp)); } catch (Exception e) { String errorMessage = String.format("Failed send the event %s exception %s", record, e); _logger.warn(errorMessage, e); throw new DatastreamRuntimeException(errorMessage, e); } if (Instant.now().isAfter(_lastFlushTime.plus(_flushInterval))) { flush(); } } EventProducer(DatastreamTask task, TransportProvider transportProvider, CheckpointProvider checkpointProvider,
      Properties config, boolean customCheckpointing); Map<Integer, String> loadCheckpoints(DatastreamTask task); @Override void send(DatastreamProducerRecord record, SendCallback sendCallback); @Override void flush(); void shutdown(boolean skipCheckpoint); @Override String toString(); static List<BrooklinMetricInfo> getMetricInfos(); }
EventProducer implements DatastreamEventProducer { @Override public void send(DatastreamProducerRecord record, SendCallback sendCallback) { try { validateEventRecord(record); try { record.serializeEvents(_datastreamTask.getDestinationSerDes()); } catch (Exception e) { if (_skipMessageOnSerializationErrors) { _logger.info(String.format("Skipping the message on serialization error as configured. " + "Datastream name: %s, Datastream task name: %s", getDatastreamName(), _datastreamTask.getDatastreamTaskName()), e); _dynamicMetricsManager.createOrUpdateCounter(MODULE, getDatastreamName(), DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); _dynamicMetricsManager.createOrUpdateCounter(MODULE, AGGREGATE, DROPPED_SENT_FROM_SERIALIZATION_ERROR, 1); return; } throw e; } String destination = record.getDestination().orElse(_datastreamTask.getDatastreamDestination().getConnectionString()); record.setEventsSendTimestamp(System.currentTimeMillis()); long recordEventsSourceTimestamp = record.getEventsSourceTimestamp(); long recordEventsSendTimestamp = record.getEventsSendTimestamp().orElse(0L); _transportProvider.send(destination, record, (metadata, exception) -> onSendCallback(metadata, exception, sendCallback, recordEventsSourceTimestamp, recordEventsSendTimestamp)); } catch (Exception e) { String errorMessage = String.format("Failed send the event %s exception %s", record, e); _logger.warn(errorMessage, e); throw new DatastreamRuntimeException(errorMessage, e); } if (Instant.now().isAfter(_lastFlushTime.plus(_flushInterval))) { flush(); } } EventProducer(DatastreamTask task, TransportProvider transportProvider, CheckpointProvider checkpointProvider,
      Properties config, boolean customCheckpointing); Map<Integer, String> loadCheckpoints(DatastreamTask task); @Override void send(DatastreamProducerRecord record, SendCallback sendCallback); @Override void flush(); void shutdown(boolean skipCheckpoint); @Override String toString(); static List<BrooklinMetricInfo> getMetricInfos(); static final String CFG_SKIP_MSG_SERIALIZATION_ERRORS; static final String DEFAULT_SKIP_MSG_SERIALIZATION_ERRORS; static final String CONFIG_FLUSH_INTERVAL_MS; static final String CONFIG_ENABLE_PER_TOPIC_METRICS; static final String DEFAULT_FLUSH_INTERVAL_MS; }