</listitem>

</varlistentry>

<varlistentry>

<term><option>-T</option></term>

<term><option>--enable-two-phase</option></term>

<listitem>

<para>

Enables <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>

commit for the subscription. When multiple databases are specified, this

option applies uniformly to all subscriptions created on those databases.

The default is <literal>false</literal>.

</para>

</listitem>

</varlistentry>

<varlistentry>

<term><option>-U <replaceable class="parameter">username</replaceable></option></term>

<term><option>--subscriber-username=<replaceable class="parameter">username</replaceable></option></term>

greater than or equal to the number of specified databases. The target

server must have <xref linkend="guc-max-worker-processes"/> configured to a

value greater than the number of specified databases. The target server

must accept local connections.

must accept local connections. If you are planning to use the

<option>--enable-two-phase</option> switch then you will also need to set

the <xref linkend="guc-max-prepared-transactions"/> appropriately.

</para>

<para>

</para>

<para>

Unless the <option>--enable-two-phase</option> switch is specified,

<application>pg_createsubscriber</application> sets up logical

replication with two-phase commit disabled. This means that any

prepared transactions will be replicated at the time

char*socket_dir;/* directory for Unix-domain socket, if any */

char*sub_port;/* subscriber port number */

constchar*sub_username;/* subscriber username */

booltwo_phase;/* enable-two-phase option */

SimpleStringListdatabase_names;/* list of database names */

SimpleStringListpub_names;/* list of publication names */

SimpleStringListsub_names;/* list of subscription names */

SimpleStringListreplslot_names;/* list of replication slot names */

intrecovery_timeout;/* stop recovery after this time */

};

{

char*dbname;/* database name */

boolmade_publication;/* publication was created */

};

{

structLogicalRepInfo*dbinfo;

booltwo_phase;/* enable-two-phase option */

};

staticvoidcleanup_objects_atexit(void);

staticvoidusage();

staticchar*get_base_conninfo(constchar*conninfo,char**dbname);

staticboolsuccess= false;

staticstructLogicalRepInfo*dbinfo;

staticstructLogicalRepInfosdbinfos;

staticintnum_dbs=0;/* number of specified databases */

staticintnum_pubs=0;/* number of specified publications */

staticintnum_subs=0;/* number of specified subscriptions */

for (inti=0;i<num_dbs;i++)

{

if (dbinfo[i].made_publication||dbinfo[i].made_replslot)

if (dbinfos.dbinfo[i].made_publication||dbinfos.dbinfo[i].made_replslot)

{

PGconn*conn;

conn=connect_database(dbinfo[i].pubconninfo, false);

conn=connect_database(dbinfos.dbinfo[i].pubconninfo, false);

if (conn!=NULL)

{

if (dbinfo[i].made_publication)

drop_publication(conn,&dbinfo[i]);

if (dbinfo[i].made_replslot)

drop_replication_slot(conn,&dbinfo[i],dbinfo[i].replslotname);

if (dbinfos.dbinfo[i].made_publication)

drop_publication(conn,&dbinfos.dbinfo[i]);

if (dbinfos.dbinfo[i].made_replslot)

drop_replication_slot(conn,&dbinfos.dbinfo[i],dbinfos.dbinfo[i].replslotname);

disconnect_database(conn, false);

}

if (dbinfo[i].made_publication)

if (dbinfos.dbinfo[i].made_publication)

{

pg_log_warning("publication \"%s\" created in database \"%s\" on primary was left behind",

dbinfo[i].pubname,dbinfo[i].dbname);

dbinfos.dbinfo[i].pubname,

dbinfos.dbinfo[i].dbname);

pg_log_warning_hint("Drop this publication before trying again.");

}

if (dbinfo[i].made_replslot)

if (dbinfos.dbinfo[i].made_replslot)

{

pg_log_warning("replication slot \"%s\" created in database \"%s\" on primary was left behind",

dbinfo[i].replslotname,dbinfo[i].dbname);

dbinfos.dbinfo[i].replslotname,

dbinfos.dbinfo[i].dbname);

pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");

}

}

printf(_(" -P, --publisher-server=CONNSTR publisher connection string\n"));

printf(_(" -s, --socketdir=DIR socket directory to use (default current dir.)\n"));

printf(_(" -t, --recovery-timeout=SECS seconds to wait for recovery to end\n"));

printf(_(" -T, --enable-two-phase enable two-phase commit for all subscriptions\n"));

printf(_(" -U, --subscriber-username=NAME user name for subscriber connection\n"));

printf(_(" -v, --verbose output verbose messages\n"));

printf(_(" --config-file=FILENAME use specified main server configuration\n"

dbinfo[i].pubname ?dbinfo[i].pubname :"(auto)",

dbinfo[i].replslotname ?dbinfo[i].replslotname :"(auto)",

dbinfo[i].pubconninfo);

pg_log_debug("subscriber(%d): subscription: %s ; connection string: %s",i,

pg_log_debug("subscriber(%d): subscription: %s ; connection string: %s, two_phase: %s",i,

dbinfo[i].subname ?dbinfo[i].subname :"(auto)",

dbinfo[i].subconninfo);

dbinfo[i].subconninfo,

dbinfos.two_phase ?"true" :"false");

if (num_pubs>0)

pubcell=pubcell->next;

failed= true;

}

if (max_prepared_transactions!=0)

if (max_prepared_transactions!=0&& !dbinfos.two_phase)

{

pg_log_warning("two_phase option will not be enabled for replication slots");

pg_log_warning_detail("Subscriptions will be created with the two_phase option disabled. "

"Prepared transactions will be replicated at COMMIT PREPARED.");

pg_log_warning_hint("You can use --enable-two-phase switch to enable two_phase.");

}

slot_name_esc=PQescapeLiteral(conn,slot_name,strlen(slot_name));

appendPQExpBuffer(str,

"SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, false, false)",

slot_name_esc);

"SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, %s, false)",

slot_name_esc,

dbinfos.two_phase ?"true" :"false");

PQfreemem(slot_name_esc);

appendPQExpBuffer(str,

"slot_name = %s, copy_data = false)",

subname_esc,pubconninfo_esc,pubname_esc,replslotname_esc);

"slot_name = %s, copy_data = false, two_phase = %s)",

subname_esc,pubconninfo_esc,pubname_esc,replslotname_esc,

dbinfos.two_phase ?"true" :"false");

PQfreemem(pubname_esc);

PQfreemem(subname_esc);

{"publisher-server",required_argument,NULL,'P'},

{"socketdir",required_argument,NULL,'s'},

{"recovery-timeout",required_argument,NULL,'t'},

{"enable-two-phase",no_argument,NULL,'T'},

{"subscriber-username",required_argument,NULL,'U'},

{"verbose",no_argument,NULL,'v'},

{"version",no_argument,NULL,'V'},

opt.socket_dir=NULL;

opt.sub_port=DEFAULT_SUB_PORT;

opt.sub_username=NULL;

opt.two_phase= false;

opt.database_names= (SimpleStringList)

{

get_restricted_token();

while ((c=getopt_long(argc,argv,"d:D:np:P:s:t:U:v",

while ((c=getopt_long(argc,argv,"d:D:np:P:s:t:TU:v",

long_options,&option_index))!=-1)

{

switch (c)

case't':

opt.recovery_timeout=atoi(optarg);

break;

case'T':

opt.two_phase= true;

break;

case'U':

opt.sub_username=pg_strdup(optarg);

break;

check_data_directory(subscriber_dir);

dbinfos.two_phase=opt.two_phase;

dbinfo=store_pub_sub_info(&opt,pub_base_conninfo,sub_base_conninfo);

dbinfos.dbinfo=store_pub_sub_info(&opt,pub_base_conninfo,sub_base_conninfo);

atexit(cleanup_objects_atexit);

pub_sysid=get_primary_sysid(dbinfo[0].pubconninfo);

pub_sysid=get_primary_sysid(dbinfos.dbinfo[0].pubconninfo);

sub_sysid=get_standby_sysid(subscriber_dir);

if (pub_sysid!=sub_sysid)

pg_fatal("subscriber data directory is not a copy of the source database cluster");

start_standby_server(&opt, true, false);

check_subscriber(dbinfo);

check_subscriber(dbinfos.dbinfo);

check_publisher(dbinfo);

check_publisher(dbinfos.dbinfo);

stop_standby_server(subscriber_dir);

consistent_lsn=setup_publisher(dbinfo);

consistent_lsn=setup_publisher(dbinfos.dbinfo);

setup_recovery(dbinfo,subscriber_dir,consistent_lsn);

setup_recovery(dbinfos.dbinfo,subscriber_dir,consistent_lsn);

start_standby_server(&opt, true, true);

wait_for_end_recovery(dbinfo[0].subconninfo,&opt);

wait_for_end_recovery(dbinfos.dbinfo[0].subconninfo,&opt);

setup_subscriber(dbinfo,consistent_lsn);

setup_subscriber(dbinfos.dbinfo,consistent_lsn);

drop_primary_replication_slot(dbinfo,primary_slot_name);

drop_primary_replication_slot(dbinfos.dbinfo,primary_slot_name);

drop_failover_replication_slots(dbinfo);

drop_failover_replication_slots(dbinfos.dbinfo);

pg_log_info("stopping the subscriber");

command_ok(

[

'pg_createsubscriber',

'--replication-slot'=>'replslot2',

'--database'=>$db1,

'--database'=>$db2,

],

'run pg_createsubscriber on node S');

$node_s->start;

is($node_s->safe_psql(

'postgres',

),

't',

'subscriptions are created with the two-phase option enabled');

$result =$node_s->safe_psql(

'postgres',qq(