GoldenGateEventProducer extends AbstractEventProducer { protected void addEventToBuffer(List<TransactionState.PerSourceTransactionalUpdate> dbUpdates, TransactionInfo ti) throws DatabusException, UnsupportedKeyException { if(dbUpdates.size() == 0) throw new DatabusException("Cannot handle empty dbUpdates"); long scn = ti.getScn(); long timestamp = ti.getTransactionTimeStampNs(); EventSourceStatistics globalStats = getSource(GLOBAL_SOURCE_ID).getStatisticsBean(); if(scn == _startPrevScn.get()) { _log.info("Skipping this transaction, EOP already send for this event"); return; } getEventBuffer().startEvents(); int eventsInTransactionCount = 0; List<EventReaderSummary> summaries = new ArrayList<EventReaderSummary>(); for (int i = 0; i < dbUpdates.size(); ++i) { GenericRecord record = null; TransactionState.PerSourceTransactionalUpdate perSourceUpdate = dbUpdates.get(i); short sourceId = (short)perSourceUpdate.getSourceId(); EventSourceStatistics perSourceStats = getSource(sourceId).getStatisticsBean(); Iterator<DbUpdateState.DBUpdateImage> dbUpdateIterator = perSourceUpdate.getDbUpdatesSet().iterator(); int eventsInDbUpdate = 0; long dbUpdatesEventsSize = 0; long startDbUpdatesMs = System.currentTimeMillis(); while(dbUpdateIterator.hasNext()) { DbUpdateState.DBUpdateImage dbUpdate = dbUpdateIterator.next(); Object keyObj = obtainKey(dbUpdate); DbusEventKey eventKey = new DbusEventKey(keyObj); PartitionFunction partitionFunction = _partitionFunctionHashMap.get((int)sourceId); short lPartitionId = partitionFunction.getPartition(eventKey); record = dbUpdate.getGenericRecord(); if (record == null) throw new DatabusException("Cannot write event to buffer because record = " + record); if(record.getSchema() == null) throw new DatabusException("The record does not have a schema (null schema)"); try { eventsInDbUpdate++; eventsInTransactionCount++; ByteArrayOutputStream bos = new ByteArrayOutputStream(); Encoder encoder = new BinaryEncoder(bos); GenericDatumWriter<GenericRecord> writer = new GenericDatumWriter<GenericRecord>( record.getSchema()); writer.write(record, encoder); byte[] serializedValue = bos.toByteArray(); SchemaId schemaId = SchemaId.createWithMd5(dbUpdate.getSchema()); DbusOpcode opCode; if(dbUpdate.getOpType() == DbUpdateState.DBUpdateImage.OpType.INSERT || dbUpdate.getOpType() == DbUpdateState.DBUpdateImage.OpType.UPDATE) { opCode = DbusOpcode.UPSERT; if(_log.isDebugEnabled()) _log.debug("The event with scn "+ scn +" is INSERT/UPDATE"); } else if(dbUpdate.getOpType() == DbUpdateState.DBUpdateImage.OpType.DELETE) { opCode = DbusOpcode.DELETE; if(_log.isDebugEnabled()) _log.debug("The event with scn "+ scn +" is DELETE"); } else { throw new DatabusException("Unknown opcode from dbUpdate for event with scn:" + scn); } DbusEventInfo dbusEventInfo = new DbusEventInfo(opCode, scn, (short)_pConfig.getId(), lPartitionId, timestamp, sourceId, schemaId.getByteArray(), serializedValue, false, false); dbusEventInfo.setReplicated(dbUpdate.isReplicated()); perSourceStats.addEventCycle(1, ti.getTransactionTimeRead(), serializedValue.length, scn); globalStats.addEventCycle(1, ti.getTransactionTimeRead(), serializedValue.length, scn); long tsEnd = System.currentTimeMillis(); perSourceStats.addTimeOfLastDBAccess(tsEnd); globalStats.addTimeOfLastDBAccess(tsEnd); getEventBuffer().appendEvent(eventKey, dbusEventInfo, _statsCollector); _rc.incrementEventCount(); dbUpdatesEventsSize += serializedValue.length; } catch (IOException io) { perSourceStats.addError(); globalStats.addEmptyEventCycle(); _log.error("Cannot create byte stream payload: " + dbUpdates.get(i).getSourceId()); } } long endDbUpdatesMs = System.currentTimeMillis(); long dbUpdatesElapsedTimeMs = endDbUpdatesMs - startDbUpdatesMs; EventReaderSummary summary = new EventReaderSummary(sourceId, _monitoredSources.get(sourceId).getSourceName(), scn, eventsInDbUpdate, dbUpdatesEventsSize,-1L , dbUpdatesElapsedTimeMs, timestamp, timestamp, -1L ); if (_eventsLog.isInfoEnabled()) { _eventsLog.info(summary.toString()); } summaries.add(summary); if(_log.isDebugEnabled()) _log.debug("There are "+ eventsInDbUpdate + " events seen in the current dbUpdate"); } ReadEventCycleSummary summary = new ReadEventCycleSummary(_pConfig.getName(), summaries, scn, -1 ); if (_eventsLog.isInfoEnabled()) { _eventsLog.info(summary.toString()); } _log.info("Writing "+ eventsInTransactionCount + " events from transaction with scn: " + scn); if(scn <= 0) throw new DatabusException("Unable to write events to buffer because of negative/zero scn: " + scn); getEventBuffer().endEvents(scn, _statsCollector); _scn.set(scn); if (getMaxScnReaderWriter() != null) { try { getMaxScnReaderWriter().saveMaxScn(_scn.get()); } catch (DatabusException e) { _log.error("Cannot save scn = " + _scn + " for physical source = " + getName(), e); } } } GoldenGateEventProducer(PhysicalSourceStaticConfig pConfig,
                                 SchemaRegistryService schemaRegistryService,
                                 DbusEventBufferAppendable dbusEventBuffer,
                                 DbusEventsStatisticsCollector statsCollector,
                                 MaxSCNReaderWriter maxScnReaderWriters); GGMonitoredSourceInfo buildGGMonitoredSourceInfo(
            LogicalSourceStaticConfig sourceConfig, PhysicalSourceStaticConfig pConfig); GGParserStatistics getParserStats(); @Override String getName(); @Override long getSCN(); @Override synchronized void start(long sinceSCN); @Override boolean isRunning(); @Override boolean isPaused(); @Override void unpause(); @Override void pause(); @Override synchronized void shutdown(); @Override synchronized void waitForShutdown(); @Override synchronized void waitForShutdown(long timeout); @Override List<? extends EventSourceStatisticsIface> getSources(); GGMonitoredSourceInfo getSource(short sourceId); static final short GLOBAL_SOURCE_ID; final Logger _log; }
@Test public void testAddEventToBuffer() throws InvalidConfigException, UnsupportedKeyException, DatabusException { long rate = 0; PhysicalSourceStaticConfig pssc = buildPssc(rate, 0L); long scn = 10; DbusEventBuffer mb = (DbusEventBuffer)createBufMult(pssc); GoldenGateEventProducer gg = new GoldenGateEventProducer(pssc, null, mb, null, null); List<TransactionState.PerSourceTransactionalUpdate> dbUpdates = new ArrayList<TransactionState.PerSourceTransactionalUpdate>( 10); int sourceId = 505; HashSet<DBUpdateImage> db = new HashSet<DBUpdateImage>(); Object key = new String("name"); Schema.Type keyType = Schema.Type.RECORD; ColumnsState.KeyPair kp = new ColumnsState.KeyPair(key, keyType); ArrayList<ColumnsState.KeyPair> keyPairs = new ArrayList<ColumnsState.KeyPair>( 1); keyPairs.add(kp); Schema s = Schema.parse(avroSchema); GenericRecord gr = new GenericData.Record(s); gr.put("name", "phani"); DBUpdateImage dbi = new DBUpdateImage(keyPairs, scn, gr, s, DbUpdateState.DBUpdateImage.OpType.INSERT,false); db.add(dbi); TransactionState.PerSourceTransactionalUpdate dbUpdate = new TransactionState.PerSourceTransactionalUpdate( sourceId, db); dbUpdates.add(dbUpdate); long timestamp = System.nanoTime(); gg.addEventToBuffer(dbUpdates, new TransactionInfo(0, 0, timestamp, scn)); Assert.assertEquals(gg.getRateControl().getNumSleeps(), 0); DbusEventIterator iter = mb.acquireIterator("test"); int count = 0; long eventTs = 0; while(iter.hasNext()) { DbusEvent e = iter.next(); if(count==1) { eventTs = e.timestampInNanos(); } count ++; } Assert.assertEquals("Event timestamp in Ns", timestamp, eventTs); Assert.assertEquals("Got events " , 3, count); return; }