Streams Implementer's Guide

A book about Oracle is like a UFO: unprobable and exceptional. I wish I know better the 80 volumes of the Oracle Database documentation. However, a few times, a few books have provided me with such a different way to look and consider things that it has transformed my way of thinking. A book can turn something you couldn't get into something obvious. And I'm sure you know what and who I'm talking about: let's call it talent.

That's why, despite the many disappointments, I continue to buy books about Oracle and continue to hope, like you can discover new genius friends, that the next book I'll buy will stay near me for a while.

So when the Streams Implementer's Guide Editor sent me a free copy of this book, UFO among UFOs, I've started to hope and wish again. And like a made on purpose, I just got my copy right now at the time I'm leaving for holidays. I don't know yet if I'll spend a few minutes or days reading it but it's probably worth to mention it. And, by the way, "How can you get to the point where you would start a book about Oracle Streams ?"

I'll share my enthusiasm when I'll be back. Nice subject and challenge at a time Oracle replication technology is for sure living one of its greatest paradoxes since 9i Release 2...

To be continued!
Read more...

My XStream FAQ

I have written 2 programs a few months back to use and experiment Oracle XStream; when I discovered Oracle had launched XStream right after closing the deal with GoldenGate (a few hours back from now...), I thought that might be clever to publish those Java programs help people set up a XStream Inbound configuration and XStream Outbound configuration. I really wish those 2 programs can help people.

However, after a few conversations on IM, Twitter (@arkzoyd), and the blog, I'm now convinced I made it wrong. I should have started by the beginning instead of pushing those programs with no explanation. I should have talked about the reasons why it's important to track message positions. How subtle XStream actually is. What benefit you could get from XStream compared to what use to exist before.

I'll try to correct the current situation and provide simple answers to the questions below:

• What is Oracle Streams?
  
• Why could you not get to the same result before XStream?
  
• What problem does it solve?
  
• Why is it cool?
  
• How does XStream relate to GoldenGate?
  
• What does Oracle want to do with XStream?
  
• How does XStream work?
  
• How can I know more about XStream?
  

If you have some high level questions about XStream, feel free to comment this post; If you have technical issues, I suggest you directly post on the Streams Forum...

What is Oracle XStream?

XStream is a new way to interact between Oracle Streams and the outside world. It uses a publish subscribe model that is different from the regular AQ queuing model. The purpose is to provide the same kind of features Oracle uses to propagate messages between databases to the outside world.

As a result, with XStream, you can now propagate messages between a database and a program written in OCI or Java/OCI. In addition, you can guaranty that:

• no messages are written on disks, everything stays in memory during the propagation to and from the program; and not only you can guaranty that behavior between the database and your program but also to the whole chain of programs from the source to the destination of your messages
  
• If something crashes whether it's a database instance or your program, you'll be able to recover all your messages, in the right order and without loosing a single one

Obviously, if you face some bottleneck/saturation, you may decide, like Streams does internally, that it's more efficient for your messages to spill out of memory. In that case, you can turn them into persistent messages or you can slow down the capture or propagation processes.

Why could you not get to the same result before XStream?

There are 2 cases, XStream addresses: (1) Inbound messages and (2) Outbound messages.

• Inbound messages

Before XStream, to send a message to a database the simplest way was to make it persistent; you could for example : "insert the message into a persistent queue" or "insert it into a table". In both cases, Oracle database guaranties you will not loose your message. However, if the database you put your messages in isn't your message destination database, you would have to pay the price of turning messages persistent in that staging area.

Obviously you could also decide to enqueue your messages into a non-persistent queue on the staging database and use a propagation job to send the message to its destination database. The problem with that method is that there is no guaranty that, if an instance crashes, the message will ever go to its destination; otherwise, non-persistent queue would be called persistent queues!

To both guaranty the message to get to its destination, without the price of turning it persistent on the staging databases, until Oracle Database 11g Release 2, you could not rely on any Oracle database built-in... But now, you have XStream!

• Outbound messages

The principle is basically the same for outbound messages. A Streams captured message could be turned into a persistent message which guaranty you will not loose your message but require you pay the price of turning captured messages persistent on the staging database.

You could also turn your messages into non-persistent messages; yes, it's possible, if say, you create an apply handler that enqueue the message in a buffered queue. In that case, there is no guaranty you will never loose any message.

Like for Inbound messages, there is no built-in feature of Oracle you could use to guaranty a message from inside a database could be send to a destination outside the database without turning it into a persistent message on the staging database.

What problem does it solve?

As you can now guess, XStream enables you to propagate Streams messages between Oracle databases and the outside world in an extremely efficient way. It allows to propagate messages in-memory only (in most cases!) and, at the same time, it guaranties messages will be delivered to their destination.

Why is it cool?

Because Streams is cool! And providing the power of Streams to the outside world is VERY cool, don't you think? ;-)

How does XStream relate to GoldenGate?

XStream, like Oracle Active Data Guard is included into the GoldenGate license. But opposite to Active Data Guard, XStream cannot be licensed separately from GoldenGate. For now that's the only link between XStream and GoldenGate, even if we can guess GoldenGate could leverage the XStream feature sooner or later.

What does Oracle want to do with XStream?

That's actually a very good question and I assume we can only guess for now. There is a bunch of things Oracle could do with XStream, like the ones below:

• leveraging the technology in GoldenGate more than just by a bundle
• leveraging the technology for existing features of the Oracle Database like the In-Memory Database Cache or Oracle Coherence
  
• leveraging the technology to build new features like file to database integration or mySQL to Oracle integration (If we consider mySQL could turn into an Oracle offer one day soon).
  
• probably many more things I cannot even think about...
  

How does XStream work?

XStream relies on a few facts:

• Messages are consumed once on the destination; whether they are made persistent or not doesn't actually matter. Once consumed a message on the destination will never be requested again.
  
• There is an order in the messages; in addition, there is a raw value that strictly increases and uniquely identifies the message positions in that order
  
• If a message is lost in the chain, that's because the chain has broken; and, in that case, all the subsequent messages are lost too.
  
• The message order is kept during the propagation from the source to the destination
  
• Messages are persistent on the source; Or, saying it in another way, if the destination doesn't get a message, the source will be able to resend it and in the same right order... Assuming it gets back the message position where to restart at!
  

As a result, an XStream program deals with its server:

• In the case of XStream Inbound, the program gets the last position from the server and then enqueue its messages from that point. On a regular basis, it gets some feedback from the server saying what messages are consumed on the destination and can release those messages from the source
• In the case of XStream Outbound, the program sends the last position of the applied messages and then dequeue messages from that point. On regular basis, it provides some feedback to the server saying what messages are consumed to allow the server to release them from its source.
  

How can I know more about XStream?

To know everything about XStream, read the following documentation:

• Oracle® Database XStream Guide
• Oracle® Database XStream Java API Reference
• Oracle GoldenGate 10.4 Licencing Guide

Read more...

XStream Outbound... A sample Java Program

Like with XStream InBound in my previous post, you'll find below a simple Java for XStream Outbound. This new program subscribes to Streams captured changes. Does it sound easy ? It is, like you'll figure out below.

Step 1: Create a Sample Schema

For this sample program, create a schema and a table :

connect / as sysdba

create user source
  identified by source
  default tablespace users
  temporary tablespace temp;

grant connect,resource to source;

col dbname new_value dbname

select value dbname
 from v$parameter
where name='db_unique_name';

prompt &&dbname

connect source/source

create table t7(
  id    number primary key,
  text1 varchar2(80),
  text2 varchar2(80));

insert into t7(id, text1, text2)
  values (1,'Text 1','Text 1');

insert into t7(id, text1, text2)
  values (2,'Text 2','Text 2');

commit;

Step 2: Create a Streams Administrator

To subscribe to Streams changes, you must create a Streams administrator:

connect / as sysdba

create tablespace streams_tbs
  datafile '/u01/app/oracle/oradata/BLACK/streams_tbs01.dbf'
  size 25M autoextend on maxsize 256M;

CREATE USER strmadmin
  IDENTIFIED BY strmadmin
  DEFAULT TABLESPACE streams_tbs
  QUOTA UNLIMITED ON streams_tbs
  temporary tablespace temp;

grant dba to strmadmin;

begin
  dbms_streams_adm.set_up_queue(
     queue_table => 'strmadmin.streams_queue_table',
     queue_name  => 'strmadmin.streams_queue');
end;
/

exec dbms_streams_auth.grant_admin_privilege('strmadmin', true);

select *
 from dba_streams_administrator;

Step 3: Create a Simple XStream Out Configuration

For our Java program to dequeue messages, you must create a XStream Outbound server:

connect strmadmin/strmadmin

begin
  dbms_xstream_adm.create_outbound(
     server_name     => 'DEMO_SERVER',
     source_database => '&&dbname',
     table_names        => 'source.t7',
     schema_names       => null,
     capture_user       => null,
     connect_user       => null,
     comment            => 'XStream OutBound Server Demonstration');
end;
/

SELECT *
 FROM DBA_XSTREAM_OUTBOUND;

select *
 from dba_apply
where purpose='XSTREAM OUT';

Step 4: Create a JAVA XStream Outbound Client

Create a XStreamOutDemo.java file, like below. Change the strings in red to meet your own configuration:

import java.sql.Connection;
import java.sql.DriverManager;

import oracle.jdbc.internal.OracleConnection;

import oracle.streams.XStreamOut;
import oracle.streams.ChunkColumnValue;
import oracle.streams.ColumnValue;
import oracle.streams.DefaultRowLCR;
import oracle.streams.LCR;
import oracle.streams.RowLCR;

public class XStreamOutDemo {

public static void main(String args[])
{
String out_url = "jdbc:oracle:oci:@arkzoyd-easyteam:1521:BLACK";

/*
* Connect to the Database
*/
Connection out_conn = null;
try
{
  DriverManager.registerDriver(new oracle.jdbc.OracleDriver());
  out_conn=DriverManager.getConnection(out_url, "strmadmin", "strmadmin");
}
catch(Exception e)
{
  System.out.println("DB Connection Failed: " + out_url);
  e.printStackTrace();
}

/*
* Get a XStream Out Handler
*/
XStreamOut xsOut=null;
byte[] lastPosition = null;

try
{
// when attach to an outbound server, client needs to tell outbound
// server the last position.
xsOut = XStreamOut.attach((OracleConnection) out_conn, "DEMO_SERVER",
                          lastPosition, XStreamOut.DEFAULT_MODE);
System.out.println("Attached to outbound server: DEMO_SERVER");
System.out.print("Last Position is: ");
if (lastPosition != null) { printHex(lastPosition); }
    else { System.out.println("NULL");}
}
catch(Exception e)
{
System.out.println("cannot attach to outbound server: DEMO_SERVER");
System.out.println(e.getMessage());
e.printStackTrace();
}

byte[] processedLowPosition = null;
try
{
while(true)
{
  // receive an LCR from outbound server
  LCR alcr = xsOut.receiveLCR(XStreamOut.DEFAULT_MODE);

  if (xsOut.getBatchStatus() == XStreamOut.EXECUTING) // batch is active
  {
    assert alcr != null;

    // also get chunk data for this LCR if any
    if (alcr instanceof RowLCR)
    {
      // receive chunk from outbound then send to inbound
      if (((RowLCR)alcr).hasChunkData())
      {
        ChunkColumnValue chunk = null;
        do
        {
          chunk = xsOut.receiveChunk(XStreamOut.DEFAULT_MODE);
        } while (!chunk.isEndOfRow());
      }
    }

    String command=alcr.getCommandType();
    if (!command.equals("COMMIT"))
      System.out.print(command+" on "+ alcr.getObjectOwner()+"."+
                      alcr.getObjectName()+"\n");
      else System.out.print(command+"\n");
    if (command.equals("INSERT") || command.equals("UPDATE")) {
      System.out.print("  -- NEW VALUES ----------\n");      
      for (int i=0;i<((RowLCR) alcr).getNewValues().length;i++) {
         System.out.print("  Column:"+
                           fixsize(((ColumnValue) ((RowLCR) alcr).
                                   getNewValues()[i]).getColumnName(),10)+
                            "Value:"+
                           fixsize(((ColumnValue) ((RowLCR) alcr).
                                   getNewValues()[i]).getColumnData().
                                   stringValue(),30)+"\n");
        }
      }
      if (command.equals("UPDATE") || command.equals("DELETE")) {
        System.out.print("  -- OLD VALUES ----------\n");
                for (int i=0;i<((RowLCR) alcr).getOldValues().length;i++) {
          System.out.print("  Column:"+
                           fixsize(((ColumnValue) ((RowLCR) alcr).
                                   getOldValues()[i]).getColumnName(),10)+
                             "Value:"+
                           fixsize(((ColumnValue) ((RowLCR) alcr).
                                   getOldValues()[i]).getColumnData().
                                   stringValue(),30)+"\n");
          }
      }
      System.out.print("  -- DML ----------\n");
              System.out.println(" "+((DefaultRowLCR) alcr).getStatement(false));
      processedLowPosition = alcr.getPosition();
      if (null != processedLowPosition)
        xsOut.setProcessedLowWatermark(processedLowPosition,
                                     XStreamOut.DEFAULT_MODE);
      System.out.print("Last Position is: ");
        printHex(processedLowPosition);
    }     else  // batch is end
    {       assert alcr == null;     }
  }
} catch(Exception e) {
  System.out.println("exception when processing LCRs");
  System.out.println(e.getMessage());
  e.printStackTrace();
}
}

public static void printHex(byte[] b) { for (int i = 0; i <>length) {
output=text.substring(0,text.length());
} else {
output=text;
for (int i=0; i<(length-text.length()); i++) {
    output=output+" ";
  }
}
return output;
}
}

To compile and execute the program, run the script below:

export CLASSPATH=.:$CLASSPATH
export CLASSPATH=$ORACLE_HOME/jdbc/lib/ojdbc6.jar:$CLASSPATH
export CLASSPATH=$ORACLE_HOME/rdbms/jlib/xstreams.jar:$CLASSPATH
export JAVA_HOME=/opt/jdk1.6.0_13
export PATH=$JAVA_HOME/bin:$PATH

javac XStreamOutDemo.java
java XStreamOutDemo

To test the program, open another session to the database and enqueue messages int the table like below:

insert into source.t7 values (3,'X','X');
commit;
update  source.t7 set text2='Y' where id=3;
commit;
delete from source.t7 where id=3;
commit;

The output looks like below:

INSERT on SOURCE.T7
-- NEW VALUES ----------
Column:ID        Value:3                       
Column:TEXT1     Value:X                       
Column:TEXT2     Value:X                       
Last Position is: 0000001d856d00000001000000010000001d856c000000010000000101
COMMIT
Last Position is: 0000001d856d00000001000000010000001d856d000000010000000101
UPDATE on SOURCE.T7
-- NEW VALUES ----------
Column:TEXT2     Value:Y                       
-- OLD VALUES ----------
Column:ID        Value:3                       
Column:TEXT2     Value:X                       
Last Position is: 0000001d856f00000001000000010000001d856e000000010000000101
COMMIT
Last Position is: 0000001d856f00000001000000010000001d856f000000010000000101
DELETE on SOURCE.T7
-- OLD VALUES ----------
Column:ID        Value:3                       
Column:TEXT1     Value:X                       
Column:TEXT2     Value:Y                       
Last Position is: 0000001d857200000001000000010000001d8570000000010000000101
COMMIT
Last Position is: 0000001d857200000001000000010000001d8572000000010000000101

To exit the program, type:

CTRL+C

Step 5: Drop the Configuration

I hope you've found this example useful. To drop the configuration, run the script below:

connect strmadmin/strmadmin

begin
 dbms_xstream_adm.drop_outbound(
   server_name=>'DEMO_SERVER');
end;
/

SELECT *
 FROM DBA_XSTREAM_OUTBOUND;

select * from dba_apply;
select * from dba_capture;
select * from dba_streams_table_rules;

connect / as sysdba

drop user source cascade;
drop user strmadmin cascade;

You're done! It's nice to enter the new Oracle Data Integration generation with GoldenGate, don't you think?
Read more...

XStream Inbound... A Sample Java Program

Oracle XStream is a new feature of Oracle Database 11g Release 2. Though it has been developed by the Streams team and is part of the database, XStream is part of GoldenGate! And now that the GoldenGate deal has been closed, 7 days before Oracle Openworld, it's easy to guess why XStream went out a few hours ago.

This post is about XStream; it provides a Java program that uses "XStream In" to enqueue messages to a buffered queue. A Streams apply process applies them to a table named SOURCE.T8.

To know everything about XStream, read the following documentation:

• Oracle® Database XStream Guide
• Oracle® Database XStream Java API Reference
• Oracle GoldenGate 10.4 Licencing Guide

Step 1: Create a Sample Schema

For this example, you'll need a schema and a table:

sqlplus / as sysdba

create user source
  identified by source
  default tablespace users
  temporary tablespace temp;

grant connect,resource to source;

col dbname new_value dbname

select value dbname
 from v$parameter
where name='db_unique_name';

prompt &&dbname

connect source/source

create table t8(
     id    number primary key,
     text1 varchar2(80),
     text2 varchar2(80));

commit;

Step 2: Create a Streams Administrator

To use XStream, you'll also need a Streams administrator and a queue to stage buffered messages :

connect / as sysdba

create tablespace streams_tbs
  datafile '/u01/app/oracle/oradata/BLACK/streams_tbs01.dbf'
  size 25M autoextend on maxsize 256M;

CREATE USER strmadmin
  IDENTIFIED BY strmadmin
  DEFAULT TABLESPACE streams_tbs
  QUOTA UNLIMITED ON streams_tbs
  temporary tablespace temp;

grant dba to strmadmin;

begin
  dbms_streams_adm.set_up_queue(
     queue_table => 'strmadmin.streams_queue_table',
     queue_name  => 'strmadmin.streams_queue');
end;
/

exec dbms_streams_auth.grant_admin_privilege('strmadmin', true);

select *
 from dba_streams_administrator;

Step 3: Create a Simple XStream In Configuration

You must create and start a XStream In configuration before you use a Java or OCI client to enqueue messages:

connect strmadmin/strmadmin

BEGIN
  DBMS_XSTREAM_ADM.CREATE_INBOUND(
     server_name => 'xin',
     queue_name  => 'xin_queue');
END;
/

SELECT *
  FROM DBA_XSTREAM_INBOUND;

set pages 1000
select *
 from dba_apply
where purpose='XSTREAM IN';

exec DBMS_APPLY_ADM.START_APPLY('xin');

Step 4: Create a JAVA XStream Inbound Client

Create a XStreamInDemo.java file that enqueue messages to the buffered queue. Change the part in red to match your environment:

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

import java.util.Date;

import oracle.jdbc.internal.OracleConnection;

import oracle.sql.CHAR;
import oracle.sql.DATE;
import oracle.streams.ColumnValue;
import oracle.streams.DefaultColumnValue;
import oracle.streams.DefaultRowLCR;
import oracle.streams.RowLCR;
import oracle.streams.XStreamIn;

public class XStreamInDemo {

public static void main(String args[])
{
String in_url = "jdbc:oracle:oci:@arkzoyd-easyteam:1521:BLACK";

/*
* Connect to the Database
*/
Connection in_conn = null;
try
{
DriverManager.registerDriver(new oracle.jdbc.OracleDriver());
in_conn=DriverManager.getConnection(in_url, "strmadmin", "strmadmin");
}
catch(Exception e)
{
System.out.println("DB Connection Failed: " + in_url);
e.printStackTrace();
}

/*
* Get a XStream In Handler
*/
XStreamIn xsIn=null;
String xsinName="XIN";
byte[] lastPosition = null;
int transaction=0;
int rank=0;
try
{
xsIn = XStreamIn.attach ((OracleConnection)in_conn, xsinName,
                        "HI2" , XStreamIn.DEFAULT_MODE);

// use last position to decide where should we start sending LCRs
System.out.println("Attached to inbound server:"+xsinName);
System.out.print("Inbound Server Last Position is: ");
lastPosition = xsIn.getLastPosition();
if (null == lastPosition)
{
  System.out.println("null");  
  transaction=1;
  rank=1;
}
else {
  printHex(lastPosition); System.out.println("");
  transaction = getTransaction(lastPosition);
  rank        = getRank(lastPosition);
  if (rank==1) rank=2; else { rank=1; transaction++; }
}
}
catch(Exception e)
{
System.out.println("cannot attach to inbound server: "+xsinName);
System.out.println(e.getMessage());
e.printStackTrace();
}  

/*
* Create a
*/

try {
DATE mydate;
DefaultRowLCR alcr;
byte[] processedLowPosition;

while(true) {
  if (rank==1) {
    mydate = new DATE();
    System.out.println("-- "               +
             Integer.toString(transaction) +
            " -------------------------");
    alcr=new DefaultRowLCR(
              "BLACK", RowLCR.INSERT, "SOURCE", "T8",
              "X."+Integer.toString(transaction), null,
              encode2bytes(transaction, rank), mydate);
    ColumnValue[] newcolumn= new ColumnValue[3];
    newcolumn[0]= new DefaultColumnValue("ID",
              new oracle.sql.NUMBER(transaction));
    newcolumn[1]= new DefaultColumnValue("TEXT1",
              new CHAR("Hello2", CHAR.DEFAULT_CHARSET));
    newcolumn[2]= new DefaultColumnValue("TEXT2",
              new CHAR("Hello2", CHAR.DEFAULT_CHARSET));
    alcr.setNewValues(newcolumn);
    xsIn.sendLCR(alcr, XStreamIn.DEFAULT_MODE);
    System.out.println("  " +alcr.getStatement(false));
    xsIn.flush(XStreamIn.DEFAULT_MODE);
    rank++;
    Thread.sleep(500);
  } else {
    mydate = new DATE();
    alcr=new DefaultRowLCR(
               "BLACK", RowLCR.COMMIT, null, null,
               "X."+Integer.toString(transaction), null,
               encode2bytes(transaction, rank), mydate);
    xsIn.sendLCR(alcr, XStreamIn.DEFAULT_MODE);
    xsIn.flush(XStreamIn.DEFAULT_MODE);  
    System.out.println("   " +alcr.getStatement(false));
    rank=1;
    transaction++;
    Thread.sleep(500);
  }
  System.out.print("   ");
  processedLowPosition =
      xsIn.getProcessedLowWatermark();
  if (processedLowPosition != null)      {
    System.out.print("processedLowPosition: ");
    printHex(processedLowPosition);
    System.out.print(" ("                                             +
                Integer.toString(getTransaction(processedLowPosition))+
                ", "                                                   +
                Integer.toString(getRank(processedLowPosition))       +
                ")");
  }
  else  {
    System.out.print("processedLowPosition: null");
  }
  lastPosition =
      xsIn.getLastPosition();
  System.out.println("");
}
} catch (Exception e) {
System.out.println("exception when processing LCRs");
System.out.println(e.getMessage());
e.printStackTrace();
}
}

public static void printHex(byte[] b)
{
for (int i = 0; i < b.length; ++i)
{
System.out.print(
  Integer.toHexString((b[i]&0xFF) | 0x100).substring(1,3));
}
}

public static byte[] encode2bytes(int transaction, int rank)
{
byte[] mybyte= new byte[5];
mybyte[0] =(byte)( transaction >> 24 );
mybyte[1] =(byte)( (transaction << 8) >> 24 );
mybyte[2] =(byte)( (transaction << 16) >> 24 );
mybyte[3] =(byte)( (transaction << 24) >> 24 );
mybyte[4] =(byte)( rank )  ;
return mybyte;
}

public static int getTransaction(byte[] mybyte)
{
int i = 0;
int pos = 0;
i += ((int) mybyte[pos++] & 0xFF) << 24;
i += ((int) mybyte[pos++] & 0xFF) << 16;
i += ((int) mybyte[pos++] & 0xFF) << 8;
i += ((int) mybyte[pos] & 0xFF);
return i;
}

public static int getRank(byte[] mybyte)
{
 int foo;
 foo =((int)mybyte[4] & 0xFF);
 return foo;
}
}

Notes:

• getLastPosition should be used once after you attach the server to get the last message handled by XStream In
• The client has to be OCI or Java/OCI. You must include the xstreams.jar library in the classpath

To compile the code, set the CLASSPATH, the PATH and the JAVA_HOME. One done, you can simply run it:

export CLASSPATH=.:$CLASSPATH
export CLASSPATH=$ORACLE_HOME/jdbc/lib/ojdbc6.jar:$CLASSPATH
export CLASSPATH=$ORACLE_HOME/rdbms/jlib/xstreams.jar:$CLASSPATH
export JAVA_HOME=/opt/jdk1.6.0_13
export PATH=$JAVA_HOME/bin:$PATH

javac XStreamInDemo.java
java XStreamInDemo

Check messages are applied to the SOURCE.T8 table

The program enqueue messages. You can verify messages are applied to the table like in the script below:

sqlplus / as sysdba

select count(*)
 from source.T8;

  COUNT(*)
----------
     1073

/

  COUNT(*)
----------
     1076

Clean Up the Environment

To clean up the environment, execute the script below:

connect strmadmin/strmadmin

exec DBMS_APPLY_ADM.STOP_APPLY('xin');

BEGIN
  DBMS_XSTREAM_ADM.DROP_INBOUND(
     server_name => 'xin');
END;
/

select *
 from DBA_XSTREAM_INBOUND;

select *
 from dba_apply
where purpose='XSTREAM IN';

connect / as sysdba

drop user source cascade;
drop user strmadmin cascade;

Read more...