@Test public void onCreateTable() { CreateTableEvent event = mock(CreateTableEvent.class); ApiaryCreateTableEvent apiaryEvent = mock(ApiaryCreateTableEvent.class); when(apiaryEvent.getDatabaseName()).thenReturn(DATABASE); when(apiaryEvent.getTableName()).thenReturn(TABLE); when(apiaryListenerEventFactory.create(event)).thenReturn(apiaryEvent); listener.onCreateTable(event); verify(kafkaMessageSender).send(any(KafkaMessage.class)); }
@Override public void onCreateTable(CreateTableEvent tableEvent) { log.debug("Create table event received"); try { kafkaMessageSender.send(withPayload(apiaryListenerEventFactory.create(tableEvent))); } catch (Exception e) { error(e); } }
KafkaMetaStoreEventListener extends MetaStoreEventListener { @Override public void onCreateTable(CreateTableEvent tableEvent) { log.debug("Create table event received"); try { kafkaMessageSender.send(withPayload(apiaryListenerEventFactory.create(tableEvent))); } catch (Exception e) { error(e); } } }
KafkaMetaStoreEventListener extends MetaStoreEventListener { @Override public void onCreateTable(CreateTableEvent tableEvent) { log.debug("Create table event received"); try { kafkaMessageSender.send(withPayload(apiaryListenerEventFactory.create(tableEvent))); } catch (Exception e) { error(e); } } KafkaMetaStoreEventListener(Configuration config); @VisibleForTesting  KafkaMetaStoreEventListener(
      Configuration config,
      ApiaryListenerEventFactory apiaryListenerEventFactory,
      MetaStoreEventSerDe eventSerDe,
      KafkaMessageSender kafkaMessageSender); }
KafkaMetaStoreEventListener extends MetaStoreEventListener { @Override public void onCreateTable(CreateTableEvent tableEvent) { log.debug("Create table event received"); try { kafkaMessageSender.send(withPayload(apiaryListenerEventFactory.create(tableEvent))); } catch (Exception e) { error(e); } } KafkaMetaStoreEventListener(Configuration config); @VisibleForTesting  KafkaMetaStoreEventListener(
      Configuration config,
      ApiaryListenerEventFactory apiaryListenerEventFactory,
      MetaStoreEventSerDe eventSerDe,
      KafkaMessageSender kafkaMessageSender); @Override void onCreateTable(CreateTableEvent tableEvent); @Override void onDropTable(DropTableEvent tableEvent); @Override void onAlterTable(AlterTableEvent tableEvent); @Override void onAddPartition(AddPartitionEvent partitionEvent); @Override void onDropPartition(DropPartitionEvent partitionEvent); @Override void onAlterPartition(AlterPartitionEvent partitionEvent); @Override void onInsert(InsertEvent insertEvent); @Override void onConfigChange(ConfigChangeEvent tableEvent); @Override void onCreateDatabase(CreateDatabaseEvent dbEvent); @Override void onDropDatabase(DropDatabaseEvent dbEvent); @Override void onLoadPartitionDone(LoadPartitionDoneEvent partSetDoneEvent); @Override void onAddIndex(AddIndexEvent indexEvent); @Override void onDropIndex(DropIndexEvent indexEvent); @Override void onAlterIndex(AlterIndexEvent indexEvent); @Override void onCreateFunction(CreateFunctionEvent fnEvent); @Override void onDropFunction(DropFunctionEvent fnEvent); }
KafkaMetaStoreEventListener extends MetaStoreEventListener { @Override public void onCreateTable(CreateTableEvent tableEvent) { log.debug("Create table event received"); try { kafkaMessageSender.send(withPayload(apiaryListenerEventFactory.create(tableEvent))); } catch (Exception e) { error(e); } } KafkaMetaStoreEventListener(Configuration config); @VisibleForTesting  KafkaMetaStoreEventListener(
      Configuration config,
      ApiaryListenerEventFactory apiaryListenerEventFactory,
      MetaStoreEventSerDe eventSerDe,
      KafkaMessageSender kafkaMessageSender); @Override void onCreateTable(CreateTableEvent tableEvent); @Override void onDropTable(DropTableEvent tableEvent); @Override void onAlterTable(AlterTableEvent tableEvent); @Override void onAddPartition(AddPartitionEvent partitionEvent); @Override void onDropPartition(DropPartitionEvent partitionEvent); @Override void onAlterPartition(AlterPartitionEvent partitionEvent); @Override void onInsert(InsertEvent insertEvent); @Override void onConfigChange(ConfigChangeEvent tableEvent); @Override void onCreateDatabase(CreateDatabaseEvent dbEvent); @Override void onDropDatabase(DropDatabaseEvent dbEvent); @Override void onLoadPartitionDone(LoadPartitionDoneEvent partSetDoneEvent); @Override void onAddIndex(AddIndexEvent indexEvent); @Override void onDropIndex(DropIndexEvent indexEvent); @Override void onAlterIndex(AlterIndexEvent indexEvent); @Override void onCreateFunction(CreateFunctionEvent fnEvent); @Override void onDropFunction(DropFunctionEvent fnEvent);  }