From 3b9ff26d1aa4cb38b938fe80874298a72da83941 Mon Sep 17 00:00:00 2001 From: agapple Date: Wed, 15 May 2019 10:20:08 +0800 Subject: [PATCH 01/17] add logo --- logo.png | Bin 0 -> 21116 bytes 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 logo.png diff --git a/logo.png b/logo.png new file mode 100644 index 0000000000000000000000000000000000000000..3e2cc642e3d659b852de136059dae9684aae12d4 GIT binary patch literal 21116 zcmeF3_g_-)8~<(E(spL1_Rh>L<=)o2)Na_;lvx$pZrulu^5=XIU;4=hcDj$Jw?ARr)g|K432 z0fGH{xBCT-9NrtYB7d{@hJ7J6CjSc54qYY+2;365fA_Xsg| zLkuLHyHNHc3YkjsOX)ScC6S_4$`cjs$OJd5fUu#hvG{F}JtVY3KE5X}A^@_6>Lylt z5ObALrn8-#B!3`1{twv5>v+|Bw`7FGQqqlq^7 zyqwpADPae&`eW4Lfsco1^Fi+YkVzTtI9z8whJJqHq=}tt7?U1Y?F}8HqPuYiR7?2k z1tsxw0FC!|RODUNZK1KlA>Mq~^oaDsPc#Rj&hu(_;Lm_UH?Ty3?A+q09EOi9>3xX22YR11j?346Os_Zi zF8_4c418|tcZt+NF?K?e8ZsTD=VzvXr*KJOhZ{VE z?(Pt|Ja8EIJoaD^WN~R6+QCzwh=B=A3M;QS8mwKiEwq%$2moEW9o>GEFtZbya`}S8 zq|}&8;>`N-=k{s7XDP(<1cfldcLVvq;cv36+?}%Dx91(u@S$*@kifMw(MF7Y(pJxR z?&m4jOO;L;IhOG(G3SmGA^C^&Ys{UZwh}JL`p7?y+b+65B>gU6IsFJ+pX%u~Cs}zq z>gUFm#pY6WgBk)je(?RO#~uC+C}}?O1{2y`a(njGzsF@v9gtnkagz@gzcA#0LW;b@ zCH4=C!d2 zCK?zJD0pQsr@?Z#9*Eap*a>Ho+k&bj$ED+1UL>|xr?#KE=vrtw43GWG_2ve1UxqAL z7wCmUFm@w#oH|Cz1n!7V(J1=1WZBsQl@gRnyi)C51^9_k=-QfZ3pGh~OvZHgi9{Xs zzzcTox0vvFVR$&P?P6Ry+{(um5qz zR#^CI7Z7>NJo8UsmvLvmtOfmDUi1%=I?b38H4(Ny)%J;kb-!$Fr*@1_wWUx$bv%kv zg0`8H?N0F{sa99gjpdqt(NTXkindsUT`pA9~Sz&us4f(4H^D%O4M zgb+(a72$H8gTGZP^51P$OC+g`8i-^C!G75rfiW6SXWKe^BVT3x6wwks{V3n{$yi@O z!~~t#;xNMwlp16@<-Q0gE4$pP$@LKl(VZjEpdu;1Z))!&73bSi&GN5B)DAM3m_!@1LfCVtp zFUOU}lmwkiieW<6u_PI1aa~V?&Xj8pb88IraLMW3?zYr=ChbSsirJHcDKB5Nj4zR; zrui)xXG78oTdTu_T{2qdVwJ(U%YYrL1SonI1g)zGa_jN$EN=)4X9SLi)uB9wht{6K zFzxvym%W(?j*)7*^aJ;GYJ}tZz|@H4=3~om$Cl+XPbd_O7vIERuwUzAu>6&Ni9HSs z!vU3w$bWKalt!RC84e-f`lC2qFs1P*K1H+C3|f_9 zAbje1+UJE@flfg04V9OS2!RLKLzA)WyeB3=4bPdIzSZb4r~YPIrk<&SM~|QExHMF* zApQ5J{?QWZfO^eyu%Ngu*$A8wcogVW73170d$uF64!ffL$>j21UuA;=8z3fAq;Nph zp*O@R(Ck4?Lc^a0iJhgj4wnq;w>cPo@iV2&ZBY?-#~^ok59wg|`?g*58Cjb7n-`3J zLTK9+ZWM0a@X{9KSakdo)lo|WHY%pXmg_-0q2%7Bb*c}qZ|V4NRR@swd5fhC3qek` z@Pq|$A7Xn(3P#9(Fc&t&J@en`KQfoCF3PLb_k6$o!-z;hHU3V3 z+k0ihZ#ezaQ|TWPnj_}-Yd$0Bf~j0vFhMJmLCew>KBFey^nPMAM)8Z?WT&e`*f!tU zVSKv45SXln<1CMpixTkdAUT$~BR%qCZ4b{7GcHkv|fPC`im}3@E-q{@NbQ&;)7;gGkaEmTe%oSfSCtrJS%pIfM6 zr;Iqrw>I!4zO0W9H;R;RL^ipOr?%{pj*xxv5jrvOT$DJVqE;Sya zZ577E9*6{s5Bj=ftL?C2V}qV{U3FMGV>bv7HbHUtpZ&(^HSmHVb(KqZ9zKa~_x`=; z?Vg_RVC)nT|RK1 zA2lMPAv8plURO!tIZ(%+8;Xd7wqC9+kOF9f+r4bf>S}k0X5xuWjbV zAWnvY0*P<_gdPalhyLT?9Kx5~L(dJ?bwLf5+^)z{ys!1{ZNMge)b^#_j z$BOW-FH2S7AI+&J2B*u6E6-Sf3v3#uS(@qn#pfS@2i1mK=QoNoe4Dz)DF5=}si-ee zvR56Ll_pUxZH{${nr&HU>n&QaaI2;s7{+z1I;1Jc;4et;Vx8nZzg0wdV{;gFadmsK zCOulMZZ={gZhg4X=U=WjDIkmUb5&7)rg00^KwwSaEVF{*>>Fc+1tgk>dA76puK)Uj z0wTB-na#cZzb(g`X&FUdGdFGUWH-T;va7 zgHmi5KyfyQrjP3H33Qb!@nBb<^MK+{;FfYkhQ()Q@#>yR_r9tT2lpJE6*K<0OK#AMXlp#JPmCr-Gb z&Z^;rrg3|tCjPwF>82*VU76677vpO(e5wr=Av4w7oo z?VPp=exJIE%e8s~VF7lyr=8&-+hHne$})0O7(AMwo-U34i4V%4Kls8S|2~VUe_3m{bVoy+I_*zQ7~2Apv8hKoM>(H8Zz>i~5mjlvVFEYy zLWEQM=Z!-f!c3{Y=kMpsCV!lcSFa$9F97H>nvWLtdC!aKC`Utm&6awoUqMk@uH`8)dc5or>P<8$IW@-y!(E zZ%@mbU-+)Y-1ivyT0C}9Az)w4RF=H{mWfYLSmqm(i%$?G=$3`Xg%5^2RojvPE^Es+ zJZ@>`$(CeNMY>>?;xE5mPgTaMpg;&c9m;&A-s@U@2r`$&5$4n~5Kp3r9=ZoshZz)2 z8-IUGh<-<5<9}P zPb7AtD)XPYbiS16RInGcG@a<;*o5`P29nb_{xLC?!a=E9ZSgkh-d0z&eV|DMNy~{{ z2ToJNN=WB-LCP@h1NRQIY}3CIE}0VghVqZAL|KEoM%{_dN+lrvd|5p=G{W9B#xT`S zOaX%F>|rPl{!AE;)UZ;RU-H)(EL-A)v3;<(;g+*y3(ZV}h6B;A_1dmfk~%;5Z5*9D zHLcN5#`t;vqk~rX>e}za+uBA^q4d~)zh<|m;2VwcL@*hTc*X+V?S5;nsR%x{2&jNg zUfM$VOZ7HZ>NmrpHbb2s_xL@)h3AL2nx1r7lO~Q5kv@)L_hv*j;NSjl(G@c93Hfn% z2wu|$@t|hf8>BmDmvEna8_--gUqFPjS2Hs&-gM05b)iz9Wh+g|7%OV0kqKNhgUNwr zFsIoDVOLe_O{0;azw$9v5muEvhD}{;)~v8`xGN%J>~2<@sU(`4u;$oX!+{xM`H#>G z(C`_e=6q-s-@#?(7eOB~FiCs5eadm)RZ~2rkO{{Q=jun#d13qTWqhd@DE;&D@Ie3g zu`6m5%^xpMHorXsqGdIVnYK0xBVhS?ICo@x9{Qn1Z#`jf;-o);Z*tZZYwyNGFIjT% z-+obc(Xos0P+T*47auzw~Oc|@Wi|c$6|rH-22~@gRCBRxHd(%>S()bHhSsr zKsT^?{m1>upOI0IC=mXy-jb8a zRT0l>p$(M5kh(koW7fKz`ZsKn`aX^9C=l~DxD6e1;z4^%7%$fRVNaAQvcWLtV#ItY zayL{`ptB*iGJpeOzI|{FlQo&rsh=U<@l)CYj2-Wn#0McfAEgN68;%NcE;DYPb)lpt zv9;{nc?b=b-&vP@!!|B$ZRqHPzC#OWw`Of$v_cAZb-Luj_gdv7h(o&_lB7Pg4#_r; zM#Lx2hu$wqweG#N7VCx9SJ~Ocbx>oTfWkL|-M_U5QPIU`0o<6jTh&hDSAIO0?M`nOWy0i z@XzR*|ERG{+rv%~mp7>{0mC^P>@)SH z8wp1hH}z!-)Y;t+R%Ak1Anz_vZ1+?2LG69%MfL_-u>6~9!JV&D=h&sb3z7EzpT`LE zPQxzIlAP;Esk@b1o}5-ca;_-_BgMo(!;!JiR`MfObBdMP*a2@8=Tna=*715ioXrHP z$h*ym+<-r48A>VX=HL#z(3~VpYj7P0$NX{;5xdfEv#(U(-8mA?srM39py{=j_>NU} z`rYiaTAUgNz@;PdIl&BAf^H}7_M?mtwA*=2_(`c+UCAd;OrRI;5{@^RzoDOHk4kYY zbzpqg^^@!O3D{+ynsUomlbI>I^$jIY{6%4OGHVYolfc(Hmntpo#*xA z1sFuItmbhh*|iK|(6qMP``&b3EKCn6C_rb8qG!K(nbZbnD|-ZFdwAfLSY{{*@z=5% zq=&-?KJ^hOlafepyKzXzc%2g#@}nG|l7lu~h$iNai=^aGFM4epD7df3s5dnPZ#h7R zx^B8O5p!$YJ#zZ?mCiffVB+~@(AwAp3cg_7chZWaXfQ-m`ni9$x9Wgn4E^-WkLM^R zf~JgD2@U14L71O-sE9Tj6DlD5}{tQy1_xGE>r{`jrNljbzKUNky|h`Y^dWw zsxuS+Tt8bc_$M@0;Nc)iavKPwh-NoxW_R!@7F?8jxM=7y-nn>xxC8rY?gY_&cnIFj zkc8TX)sv#8{2=f&WW30|P7+q+3B}t+hVL%#Saa*Kz&oyy4Y8pPP-2 z9MEFRm*9e1VLPoeqU-yd{~Cifk~+?3Wyiof#=!94;XbUvY+MkdRR8YhS>yLpHkm}G z(`J7$T4(+P@3rv1*UW?nsG8V;IkzbAJVIv0o>=j7=ZLEA49-o}3=c{T4*&SfcnT9> z;kVob?chMG^mo0_U>=(9T#=&YIXp(*z%;9z?1v6ZG_zTm zh5lK39po9TWcyE@?-{jfV?rTW=q&88r9oh0#}HbVriYabPIimT-fzoe5e{rmpD#VC zmo*9R22phSdnMY}p7$cg3#QN4^5+kF8aADfzHfG9+Oq^5#kHypU$uzL{MP+D{Pv; zFVoOn-(M|M8&Fr2l?tFDVQ#pdfgN(n--1cY}}lYxLJs;air`1s2P~f0G}D zHZ$t4BlusJ#z2xv3yWLP!lfd)++T@Hz4^6cj!aY=ySIUq{os%;uiBso_c0GDGY{`J zRE}`)^^TuF#qWw-0^KOG$Er=tPMP%@eAA=bGAyyzt^9R@{munTzIYUR=$jUjI{Ev1 zcie3gNpPl>^OF4j2!879sWOtl9|q$+TpfgZC>=vuIb8HCTUmFvv_Vl;{nG;;knIxX z*b;H7t?dp|bhQV5fK3^!8TmB}eW;%LZ{F{SKNP9o4Y~wNySH{W)TW?^zkRSDmhDZj z!b^ZGh0dj%t&T|{^X0b~{7T2@*Z?bQKByL@y&g5V&8|ys0*B#7(>N&$!+iUW9=SMn zpoP&B^gx{%H?FHz?+l2)_x?iXvBdRbOlRuTjT3@R>MExob z)~$W5G`pleKkB7W=aYwadlm+H#ZB2!C&j3jKH9ADNseKGG`nU}qW<0v6!9xgy&X-2 zxZ9~$=u}hl2j=-cJJ|)OnMWkLq)dxtZZ)m}z$f`8wZtWU;isg`MQmK~A7Vgc3aE#k z{W;?2y+Gawi6Qc8>i8S^c@sPfxc*PzypOKHju1$S(p((y-D0n) zl<5Ir>np}ZC9_P_4ql-$NJt?xQrX>?lfG^kD5}vdbSVfM7#r=<18`TSY0*Q7&XaA1 zD5T7G6EydB=tA3BL@es^sny8B&#WrW)Pve%KQc4*|))eGyV0DDFr z5EK;bcUm-c@)Rmb>Zi{`WkxY6e&Zp@%W@V6U)mqHtJ=XEb{t)g?BKAxs`7~+Ud4tx zSZtqkn^6gmWi8-VLYO=Wb}ltS(z33ZYr2hY)_Sb#zx>n>;R)g`cuCsKY}W(Nd-F%z zIN8X6dz57)HZXkG)X}kadgsWMN92j(N91@R`zU^I!H)yD%l61|+g5%7QgfI?M$;`LYqB7fE4C$QY={qIA`M35ZQ1yh z$t1b0uO2eM_RP$TdN;LS9`1zmw6UkLB@09GR^+x$_GrBeg~My!_Ry00fi+-#+bkfneolz>>>Ey*Ib2MZ{-`&hiiZK~Oe7Z$i~gf0 z{vW=8&cb)-F>Nhp0lLqPGKgm;orG?0Qi{xByQ5Bl%@}?}OH;sPJ!(5hjXCf7R`?*U zDMyEGPuf;;uv5nnELxc^}#2fN5g^{~ZTuV!S0(yphdjj_h2)u$AblK z6l}G92w-#y&Fb!iyB)3iC*#?Se~xnSB@<6kPNRKAEHagW31OoCK41MLHl|Yq=DcL2 zVKY$>4+~m}=Mb-N(O$?~F)(3PQ9 zjV6C4iLJ^P3f<$?S*a8jWsMg;r}kzz){mRW%;J{R!hTJ14ivP1iF*1J`5%%}ee@k- zwVS>5{xs%t8kbaH#-G)pjxR;={+zH7Jm6&_DR_9ai$spS@x$(iQ@iAqA9kr7d_nW7 zBjTI`FU;0t_V9?eIyQPlBV~>VQ%g2;Qb3086ca0nl#7lHH4g4~7xXoyBwjcAcHQp6 z$xH0bZqi*S78mtke!9p9&4I=S{wtV|nd4GH+UCUFhTw`@GFZ)Ngu8IE*(k!BUzMxu zO&#?ixF~f+5a6CuDEG6JxL2(_kq$c3 zk}`Kg&v+d7=y|7!auhK&Hapx}NbCu%E1Du>C9`orQS9*3Hwup|)?X?iM}yu*4HZnu zICCJsj(ImN`n%-UH1T3M=V8vIr11A}4u&r`A3~cUs&-F$*?_b!YoL=a2kE{`GhBBe zH!Q)U5-o3;!o(pF=9#JcI`1Hgliv;2wg0AyLoc6-tX|wjBYzXw?k22#RR!+1>{VQf~Bs6+NcG>eDB8*H~Izew$H@aZUY%0z89 zV#XJg-?{eun6nz=KNe^C1T%xE2dQYTG4HUU90c$?mPP#={`-kg0Vjq1 zIE1t-=%i6xhIGa?RWkBNx}p9iTJyuaf)T<71|l<$e*S4p)v+Z{w)xh#_Nijt58A8! zcdnCZ{2rVUOuTa&8?0LRYOfd%V^WC}OLxHom2<$1C(v;8H^g>62y(ySM3smZTu=aC zcDwlekDp3dcTXKk)-X35%4|FR%-BiL@6o~x1kBz& zVx(8a-w!NNEesTN&5yRyQ-D-rT{@~LUr1}Bu0MyU(e*m$w(Z8h;{30u=hJxu?22E) zH+oUKK_Ek1prF5q?{@8a)JRdUDD?7KStXm?Tmctz)A3AxY3Cs_P6WkSP;m)(7T$8G zSJ~U}SQ;=0#BUPncsSAw-&))WpIdAWKZFcLa2K};V;!?;xtjh#_4i{AlM0!oo!$F` zn$YMPOUnpL%>1giyE}C6{hUidutBrVCR?|q?EDc$z^5~aFix>v?-_s(1aY6C=Vtb#g64wv84Nk^s6QHNo!w0tDH7v z5`3FE71&YT4w=)T=9-~>sX6KWmN6r3y%GE+i5HXD{_QP|03kip`9kS;-3O@5fCCDH z1_F`PR*W$#BV~TyH_|vg;J~h3Q`2@02^&>UckFl&BkDRLXyNy?OsF3ObD$?N%Y*MA z4XEM_;z@CHv`SiL?p(S(zVin06?^bQluN{dz}GcY#$_iP*Io5_Xl2K8Nq= z9{c&8YQ@6ySx3bqXAw8v=4Y-c#D|2;L-7x=zu(7UTMZqUHA?{O< z<^wgELv;mv%-|$=^v3E}b^Abpq;+?n3;nt~AAv)=$gkJ+{^5>00-Yez*CWOENO0`I zQ<`hb6oXLd{vps3txTm&X~$_Z&f}@z&AvwMe^2V`WlKJ6*#2v>4a|=XnGNU7R>J9g zL|)LmJOyqt6dUk6s(`c3R%?nHEyp{2;CDm=a&dKrtkjKb(JRrtan9?`fw}`$Yt~ct zY`V`JD{(4cSvyTkBAUBN1%TNyq_boN96+I9-7j8Jm%G{-_IAMlvrXECTc#P(er(=a zC#x5s<@htClw9uJDqRgh0q7IZaXInV;ka4UB}BIdaTAQy{2LC$J3Fo960PE)2Ib^j zpZemR`))ky@W9sv&>()Jr5>>|%G&p0qOCHQ&oo>%G%kkY7#k=b##91t<3ZGWq>>de zs-$KJe{x;s*+du9S}>}Kc(NS`Q(hZiLe}K3Q=95jwU}=#jjQb5|~nY=~@aDtcYBEV0>Os%`hOvCt>;AL8yNR33RE8F|d_P}KARxIM;dv{3&B`2%Gvw~Uz$Od`9*N6)(c>a2 zUG;Ri;deWmg9McGtZ$xJI(|8K4_)ccm+qTRkR^-hq_*H8B*f1P{)^{;Zn34gZ>uV& zI1eNT&~+o53DehE^pq-cbvrMH9g%16AYVD&3*+>(eGJ&#`5ZBs+RXGxHKfGscz0Yz zmeE~wD>phLCS+!#n>ht~mc*af1t$n}`VmuR>&I;@z#=4?x_b*V84(+f*@>7g;7D`W z!4vd&kdL>ddk4~`RezNX^1;WxXgvEs)GK-Ets77@dTw`$DW=)U6cp$nfBXPZ`XSOu z{-Zt`C?Z`g1pF>dwF#DAFoPEGzOn@miIS65Ut;*ox1FTrJmS*$svk%ftMeX&ZCYXG zPgGNPyLk-E{kg}w6W)9dlhZ4a$6tBLIBCB&hcC=u6HFvL~}C#%lZ6mu|(fKb zz>c5{1wTjr+hQotDuetECmh8I&$7LGYa4iK%@OR=s#;fM!KSo-*tC4xsXfks`-*+k>rIm)^+wwOGo`=t?Et z7a5WnfBp6lP?J`1w{w&|lhJ}+Z6B&#Ymy`&M`k%k-6tH^8X(U#VSHQaBgO(s$`L+7h4;!`Y_^EHnY#r3VvdF-yu!~^^lUAWnIw-T2vy)pHSFD_fiuJKz z>{?^Q)uEZF#yR&56D7mHC&K@*WU)Tsyc$8>ac(X?|E32v_Kar7^$u2$FSbMcA#d0{ z$kMOrU*0c^z8DaHGFvBII94b>| zPqfkNF8g)|6j<1@(!MN!KJFT?Sz=-K%rw^pJBk=$Z0%cT3zT2#SljRNW@wpF*8MW( zN`2wkZ8fi;h;{k&s``=1_6T=Xqr$TK6}-8PZ@Xdv=`U zsvjYIuSeb4K{XMxtM?5?H_g{$EP!(vPf?vHk zaV?)4Bx5uI{~OUI38}bV&^_5f&mAKpx#_+BW|Cs?t%x~ds`KP~BSM_M-BT3ZRJZ(D zEZT6Pm|h2){l9s9a^o}{fOgueeQ1!GztQ~KVdhfO!bw5XoF{+&u5J}~F6wmw*Z8p{ z)J(C-R2OCZB`#bPLb4O8Jrxc*@mT`&cWA#g7F;UTF8T3YoNQ^Q4$ zYW`u{f&nJgR!{bQfFeoq16NqYFwO z2;D4Coo(?hLVr_s#Wg=OHm`};=VwO8+D6R{I@emYwK7*1RghC<7dc(XGf0^VIcp2dbp2xPr)&{r>ozexa&8zOZ-;NH_I|>9 zxSie0)D+S|d&peOik!gndoCRUJH?FJh4mK~?Kusxct7a(KY!%nqys8wz`qZpqc24J z41XCg_VK-ndl*0l(`j;ZK)YQ3*{1KKDEo%M7uHEaz}p8Y=2A4-*|8dHi2v_Av~SLI z(}?FyuzvXdg1nVH{t#lqD%&gAJowqK9>6fm(#=qLUINPJ~asP zSMQEehAWKDg@JMm(1ga#C($cqYrTUd?W~LC^D;m0kKVc=d`!(S(QoJz*#^x`tlGbM z`W38v*KY}6#(ei{Y}DA$Wo0A*A#8_7_LsYdTs$HFC@z&OouW${GVE1Ai<5=ky8R4k z!MF-DrW!o^W^#1|d-Sb4z+YI7006+X)gBBH-%E&5JAY_IUmcEP#vGfML3mrn(9 ztmJF+d;6Wn#BLg`{AYO4!q@cMxV4zX_0dQB&S1V|I$3Z&5G3&vks`3v&p@23zw*q$ zwbIPx0x%ft*NL;G1qSqJCpO7;%K%m-!^K9>Rf|Jrj5cXtr(Uw2i8)g3fpsO z);G2%*5boXJtQ*AP%Ziy_pJcgE~5P^fVAbd&Q56!;xSp9L-c#}q^);iitj84YGP=9 zX+d2IYmnp@a_G+o?zI|4w>tH#>BX*_N{!1(#IuS8nzrLkp$CaVgIPt2mS?Ac&7Dcs z2G?VjTs=452A;Tm#=qT=B{AwRgWNQd{gL8@Bw)g4I6tp7p&%4B{J5E3icoD7^Jl zwB&MHw{E-p<+7rIa${E>5oOu`7bwsr|w4JDJBiR!${Ik||J0&5hMEiL>GqDf zriw)R!%zxp%2rd7zqnR@0oZo6z$T=(-16pj;%Sn2Ny*Qb-Gzn>n#Cq_z#hn`GEb7K z=4q}h{$h>|Iezc23I((m#wf;(9mXp?24~c`!U#EGRQ|v3iVXtxzXs^lf(Fz@&tMP7SHa&jzGhGd2i9 zePK^FlIE#A?`N>Qg~JxjoaFYla}+eB2&f1>_)tG5e0^>E(Z$y>GUe4-y%8VRWRM58 z?orR4FNz&C1EYQau~@FQ_HIgK-1i!Pu(nMDzBF11R#h0T=+yp$S9Gn4eUtlK+qJ50 z2&30`7Q!mD#4Fu7+IC-E9L8_)Pe<>6)&_AzSjwC~vM{1FeH|rURJY{G8`sb=M9CbIyjWfM+0h5H`gl* z!aTWmNehcr@7ttoHJ!tUM+soS%s?gia`ASP8uf)B?MCfDIwI;3~%j(!3A`T0;2eUnvj+k@y3}$R!f+l|k z>oElX6J~-|J=H@tZr~i+1*X@a zJq+&Os6{%{k>FB0bEd2JDq=1|Ke9#5KxjXy!N)D+5oTv;fi;N8$Ff73j z<`WO=?~t`j2i~O*(3P6DlY>SF-vYAjzmZ1q$`%>Pmp$t4ZDu)LvI47SJkK9LGx5xG z(jMuBS)(%HLSSgN?h*s=&|=9%X8m44{-k=tLh-_*h*VNV{Jij<1DU({H#Nnjf@}QE zn*PC~{6$M0INE=e^Y+O_>7vH6)RmMWjwN(dxq-Y|7OVBUQnL1_F&z zg$COfOH^t0_nuhG#@Wq}tFjIV-26=h{o4&Z9{SdAF|qOg`Em4K(r_RA4Q_U*&_6l) z+`SoC;^VC5^beqnc46ai%~K`rF*0x=;)vazx||gxIJGi0^~fezOGZi%4Szg z_RPe6{GDwC6$$&PJj%I4CG2{Qk{6eI|7E=f3kaN%|LOA=)C+-HlTvB;Az9bhLhKWacb9Jo@OR>HVt4 zN`CCNYGno6Xre`1pk4;6{Wej!Uezkt%^JCKgaeL<1@Pc z_UWFkc%*5xs9Qq9-QjG_R@ipbml)eJ=}7aA&ND52)-UTS1Kv%{bE)di?qRd(aY-64 zZlN{jBly@mGdk$qprDNSwe+!ghej3|8G<#XrFr}8BSq=7or63bQszgWEFCfzFXZ)# zqI9&Vi=D-`ePuB+qW|E>|A?2xZ+#&LUwP1PlGhfQv3=}07$BjptSA|lsRcTbJEE8S zO!1aEq33}G&Pd@ym3=IO-=Ckj-292A0t^B^IZnuQ?2ALnwX=pr`hj}~(>D~e{#NbN z%#x8-rK)FdgPO5K@yt{Uk*2Z3!yG)*20ZOOZ#i5;ct<;aK|?dc&F8nIbk6YW=Z9Rx ztDaXMcrlJBaUDp)d@=DyhKX*@BVi=~bJ9mF9;*L>@Vyo`Ed0o@m4K+?DpYK@)(};h z`JxQ4vqD8wH=%*I@^@`{0y@`vq)(D(8=i=wBZh`Lf*9HP&UKq*s5!)R}}(G$TRd9#j$aF%aH!1a?Pp0U}@CA9l4tY_Y_$6EdcaYwIb_|&Qg5` zK$>bPwodv^O4B%hM$LUsxL^B)2yE1@UC135d_;PDalz(7;8&41nE{2DS*D^TnHySo z*NMo?L{fsoO7h+NB@H{z@vM`Vn#Rly`7S$rb9MVcZ28*ji{FtA!FC5#tp8T24j%!4 z8EZrya%Z9zdtGtcz)yiPX=s0t3t=9QY0)o0E)`9=j?VK9rU0E4?Y%D+`ycgU8va&Q zjhXD_oaE@Y`I{AN(MgDYTj*~K(5idxP!>j-zDioY1nWDGF64*MRwd1vQky8B71+_4 zsiub^4OY>crBH^PYyBMLtR4_=Wl+wEMs zW?a;>6aI9Ji-(fLjFH6PQW&pTDVJ}IS@e1wCHgW&f~)%V?O+nPf(du*;Isw&TFgUxL*ipMMOX2KI zeG`tU#8OJT5Mf9~u{HzJ^L6OJWUYj<{Y)x9BVC5!GkP*BFD&$ViBxjF4ZMGH*%~Y; znKAX1;9wlqX^@9;{S2E*lxbc|{xnzTdzH}e!Eq@aJQQ~gQ1a_Dc{~yHCFYzj&AISL zuS$j8@+M9DX~<`6e$*78Hsn(2)VjbeKhDsa-wIvvc+dD?Gv&42K!0eo8$3wY>b(MM z*5?>3GTCg;e|yIJ*QJ!FW^FO&QV#=ZcaQ2$Zv80l8&@pXymR02^L5N=k8FtINMTH! zBX_>buNQjJBtN*Z?g6q#r>Cl4Z7J;f4ZP&i<>>;T*B4h^NAbTjXfzJtp0A8hxXv#T zH;Ht9iJt6zgIRI9y$FDQF(_N0@?*9VM~P8<0yjT=ZFHY8>>{#2_*6zO$B6XFEXY~- zoGGYctF$vqIw9Ikt}eDY10X12t<0Wxs{L$&r&vk#L~2J{#|p)v+Lx`ra^;YZeNE(w zzE%J^wi=+=H<{D&K?N%ic4?l%j&IeVXHEPb-x`@kcCcYvWfK^(@3%K^&uQu$ZPGtt zbw^Zz{73u04iI^aW={w>NKM*?zV*-5=3v3N9wPeugGw1ZvCmV;eV=rt4TqZi=oB%y z?Qi&VrX)RiO5f3rA3w<4GpA33=sr_sk+l6x?HxN14*|!q_8KapV?wB2Rna;*$oMr! zx2#yqT4Wq@xb-Mg)}lF|xU$ZVoqz^xfv!`Lc|X@C{V#XvQKX=9x_y0*v_lpy;zLkI z!6goZ|3b=3GmY+Sp)P>fE1y55hvau^0S&(A2LehY&Tl_cp3ZRmcILve0)5aY=f15$ zT`K^_j;iYet_2RL*VV@FpRh2Ne&362-kKa`%1decU8^O2Km z%lzk9RQ61Y0pGEu#)Eo(0u(rtmH!CCqy}snJYDJ_2g%32_|o#}Aa*a8O>pmGx#;@l zeAlh58Lo%N0H-jIDIwLvCK2}bP-^Bx2f2SLz4}fxvW0Fnt%CD}&Pk5UI<3&+jsq+- z;a?oiH^){rhCfT&2H|=$g;m+NDFQ7`4ht8#CEFjt!2YVPG5hcja*vdh`w3q^BZv_J z(lJ@mz@;v7u_dNB%FI9#He+mKw@z2)A{@1xE|=g#GaWO%HdYgK?EFq?OqHDh~KFFGg*q5u2?oSSr{U@Puo0q^%k z{-WUt+W*;~hw^R_>H21;O{L3j_HVCwqWkT`G)dtt3(GC0Ex!I~_Q^u8`6vB;pEe7) z^X*{OnJ2&*&YYWS)4n(6+~}KqRp0qZqW3+s=v?b5JF5NVTDIGXJj`9r`>3xvLBQ_g zM!|p2YMXaVo@c``zs|p<{^KIYJMz!Af8Fuz6-#_^?W?=RRjVL-3uc+l-5pf=U`dGV z9{tdEKaZ=P#cq4<+A-9e__yummwyuH?dPoexpw8PYyNwGHjCYxwCbPYZ);WA7mwb( zNxJ#M#3T_k{bUVfW+ZNm20uSH(R|-z8{VUAgLNNap@`RTWYC|H}_Ji&E|Mgwf+z zMvr3|J&tAcIF``@Wwaz5EeS_-NocU|z<>TT^R5)h9d)<_JWi6q)78&qol`;+00SSu A^Z)<= literal 0 HcmV?d00001 From d0a83aad2758c28e84b5581fbf740ed5a4ffee55 Mon Sep 17 00:00:00 2001 From: Heng Du Date: Mon, 27 May 2019 23:51:32 +0800 Subject: [PATCH 02/17] Support seamless docking with Alibaba Cloud RocketMQ commercial service (#1849) --- .../adapter/support/CanalClientConfig.java | 41 +++++++++- client-adapter/launcher/pom.xml | 1 - .../launcher/loader/CanalAdapterLoader.java | 6 +- .../loader/CanalAdapterRocketMQWorker.java | 24 ++++++ .../src/main/resources/application.yml | 7 +- client/pom.xml | 4 +- .../rocketmq/RocketMQCanalConnector.java | 46 +++++++++-- .../otter/canal/deployer/CanalConstants.java | 6 ++ .../otter/canal/deployer/CanalStater.java | 20 +++++ deployer/src/main/resources/canal.properties | 6 ++ .../rocketmq/AbstractRocektMQTest.java | 6 +- .../rocketmq/CanalRocketMQClientExample.java | 25 ++++-- pom.xml | 11 +-- server/pom.xml | 8 +- .../otter/canal/common/MQProperties.java | 65 ++++++++++++++- .../canal/rocketmq/CanalRocketMQProducer.java | 82 ++++++++----------- 16 files changed, 276 insertions(+), 82 deletions(-) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java index 3574b678..00939c58 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -33,7 +33,14 @@ public class CanalClientConfig { // aliyun ak/sk private String accessKey; private String secretKey; - + // 是否启用消息轨迹 + private boolean enableMessageTrace; + // 在使用阿里云商业化mq服务时，如果想使用云上消息轨迹功能，请设置此配置为true + private String accessChannel; + // 用于使用开源RocketMQ时，设置自定义的消息轨迹topic + private String customizedTraceTopic; + // 开源RocketMQ命名空间 + private String namespace; // canal adapters 配置 private List canalAdapters; @@ -133,6 +140,38 @@ public class CanalClientConfig { this.canalAdapters = canalAdapters; } + public boolean isEnableMessageTrace() { + return enableMessageTrace; + } + + public void setEnableMessageTrace(boolean enableMessageTrace) { + this.enableMessageTrace = enableMessageTrace; + } + + public String getAccessChannel() { + return accessChannel; + } + + public void setAccessChannel(String accessChannel) { + this.accessChannel = accessChannel; + } + + public String getCustomizedTraceTopic() { + return customizedTraceTopic; + } + + public void setCustomizedTraceTopic(String customizedTraceTopic) { + this.customizedTraceTopic = customizedTraceTopic; + } + + public String getNamespace() { + return namespace; + } + + public void setNamespace(String namespace) { + this.namespace = namespace; + } + public static class CanalAdapter { private String instance; // 实例名 diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index 1aa095b5..27582d5f 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -62,7 +62,6 @@ org.apache.rocketmq rocketmq-client - 4.3.0 diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index 9974928d..7623c048 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -130,7 +130,11 @@ public class CanalAdapterLoader { canalOuterAdapterGroups, canalClientConfig.getAccessKey(), canalClientConfig.getSecretKey(), - canalClientConfig.getFlatMessage()); + canalClientConfig.getFlatMessage(), + canalClientConfig.isEnableMessageTrace(), + canalClientConfig.getCustomizedTraceTopic(), + canalClientConfig.getAccessChannel(), + canalClientConfig.getNamespace()); canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); rocketMQWorker.start(); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java index 5da53763..3193553c 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java @@ -40,6 +40,30 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); } + public CanalAdapterRocketMQWorker(CanalClientConfig canalClientConfig, String nameServers, String topic, + String groupId, List> canalOuterAdapters, String accessKey, + String secretKey, boolean flatMessage, boolean enableMessageTrace, + String customizedTraceTopic, String accessChannel, String namespace) { + super(canalOuterAdapters); + this.canalClientConfig = canalClientConfig; + this.topic = topic; + this.flatMessage = flatMessage; + super.canalDestination = topic; + super.groupId = groupId; + this.connector = new RocketMQCanalConnector(nameServers, + topic, + groupId, + accessKey, + secretKey, + canalClientConfig.getBatchSize(), + flatMessage, + enableMessageTrace, + customizedTraceTopic, + accessChannel, + namespace); + logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); + } + @Override protected void process() { while (!running) { diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index b53e8735..14bc3e6e 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -10,7 +10,7 @@ canal.conf: mode: tcp # kafka rocketMQ canalServerHost: 127.0.0.1:11111 # zookeeperHosts: slave1:2181 -# mqServers: 127.0.0.1:9092 #or rocketmq +# mqServers: 127.0.0.1:9092 #or rocketmq nameservers # flatMessage: true batchSize: 500 syncBatchSize: 1000 @@ -18,6 +18,11 @@ canal.conf: timeout: accessKey: secretKey: +# enableMessageTrace: +# accessChannel: +# customizedTraceTopic: +# namespace: + # srcDataSources: # defaultDS: # url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true diff --git a/client/pom.xml b/client/pom.xml index eeb62ea0..13083413 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -106,8 +106,8 @@ rocketmq-client - com.aliyun.openservices - aliware-apache-rocketmq-cloud + org.apache.rocketmq + rocketmq-acl diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java index a39c1ba5..a1a6d83b 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java @@ -6,6 +6,9 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import org.apache.commons.lang.StringUtils; +import org.apache.rocketmq.acl.common.AclClientRPCHook; +import org.apache.rocketmq.acl.common.SessionCredentials; +import org.apache.rocketmq.client.AccessChannel; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; @@ -24,8 +27,6 @@ import com.alibaba.otter.canal.client.impl.SimpleCanalConnector; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.exception.CanalClientException; -import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials; -import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook; import com.google.common.collect.Lists; /** @@ -40,7 +41,8 @@ import com.google.common.collect.Lists; */ public class RocketMQCanalConnector implements CanalMQConnector { - private static final Logger logger = LoggerFactory.getLogger(RocketMQCanalConnector.class); + private static final Logger logger = LoggerFactory.getLogger(RocketMQCanalConnector.class); + private static final String CLOUD_ACCESS_CHANNEL = "cloud"; private String nameServer; private String topic; @@ -54,6 +56,26 @@ public class RocketMQCanalConnector implements CanalMQConnector { private volatile ConsumerBatchMessage lastGetBatchMessage = null; private String accessKey; private String secretKey; + private String customizedTraceTopic; + private boolean enableMessageTrace = false; + private String accessChannel; + private String namespace; + + public RocketMQCanalConnector(String nameServer, String topic, String groupName, String accessKey, + String secretKey, Integer batchSize, boolean flatMessage, boolean enableMessageTrace, + String customizedTraceTopic, String accessChannel, String namespace) { + this(nameServer, topic, groupName, accessKey, secretKey, batchSize, flatMessage, enableMessageTrace, customizedTraceTopic, accessChannel); + this.namespace = namespace; + } + + public RocketMQCanalConnector(String nameServer, String topic, String groupName, String accessKey, + String secretKey, Integer batchSize, boolean flatMessage, boolean enableMessageTrace, + String customizedTraceTopic, String accessChannel) { + this(nameServer, topic, groupName, accessKey, secretKey, batchSize, flatMessage); + this.enableMessageTrace = enableMessageTrace; + this.customizedTraceTopic = customizedTraceTopic; + this.accessChannel = accessChannel; + } public RocketMQCanalConnector(String nameServer, String topic, String groupName, Integer batchSize, boolean flatMessage){ @@ -73,16 +95,24 @@ public class RocketMQCanalConnector implements CanalMQConnector { } public void connect() throws CanalClientException { - RPCHook rpcHook = null; if (null != accessKey && accessKey.length() > 0 && null != secretKey && secretKey.length() > 0) { SessionCredentials sessionCredentials = new SessionCredentials(); sessionCredentials.setAccessKey(accessKey); sessionCredentials.setSecretKey(secretKey); - rpcHook = new ClientRPCHook(sessionCredentials); + rpcHook = new AclClientRPCHook(sessionCredentials); } - rocketMQConsumer = new DefaultMQPushConsumer(groupName, rpcHook, new AllocateMessageQueueAveragely()); + + rocketMQConsumer = new DefaultMQPushConsumer(groupName, rpcHook, new AllocateMessageQueueAveragely(), enableMessageTrace, customizedTraceTopic); rocketMQConsumer.setVipChannelEnabled(false); + if (CLOUD_ACCESS_CHANNEL.equals(this.accessChannel)) { + rocketMQConsumer.setAccessChannel(AccessChannel.CLOUD); + } + + if (!StringUtils.isEmpty(this.namespace)) { + rocketMQConsumer.setNamespace(this.namespace); + } + if (!StringUtils.isBlank(nameServer)) { rocketMQConsumer.setNamesrvAddr(nameServer); } @@ -131,7 +161,9 @@ public class RocketMQCanalConnector implements CanalMQConnector { } private boolean process(List messageExts) { - logger.info("Get Message:{}", messageExts); + if (logger.isDebugEnabled()) { + logger.debug("Get Message: {}", messageExts); + } List messageList = Lists.newArrayList(); for (MessageExt messageExt : messageExts) { byte[] data = messageExt.getBody(); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java index 45cffa5a..78f22e72 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java @@ -53,6 +53,12 @@ public class CanalConstants { public static final String CANAL_ALIYUN_ACCESSKEY = ROOT + "." + "aliyun.accessKey"; public static final String CANAL_ALIYUN_SECRETKEY = ROOT + "." + "aliyun.secretKey"; public static final String CANAL_MQ_PROPERTIES = ROOT + "." + "mq.properties"; + public static final String CANAL_MQ_ENABLE_MESSAGE_TRACE = ROOT + "." + "mq.enableMessageTrace"; + public static final String CANAL_MQ_ACCESS_CHANNEL = ROOT + "." + "mq.accessChannel"; + public static final String CANAL_MQ_CUSTOMIZED_TRACE_TOPIC = ROOT + "." + "mq.customizedTraceTopic"; + public static final String CANAL_MQ_NAMESPACE = ROOT + "." + "mq.namespace"; + + public static String getInstanceModeKey(String destination) { return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java index 3a465bb0..41595461 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java @@ -207,6 +207,26 @@ public class CanalStater { mqProperties.setProducerGroup(producerGroup); } + String enableMessageTrace = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_ENABLE_MESSAGE_TRACE); + if (!StringUtils.isEmpty(enableMessageTrace)) { + mqProperties.setEnableMessageTrace(Boolean.valueOf(enableMessageTrace)); + } + + String accessChannel = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_ACCESS_CHANNEL); + if (!StringUtils.isEmpty(accessChannel)) { + mqProperties.setAccessChannel(accessChannel); + } + + String customizedTraceTopic = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CUSTOMIZED_TRACE_TOPIC); + if (!StringUtils.isEmpty(customizedTraceTopic)) { + mqProperties.setCustomizedTraceTopic(customizedTraceTopic); + } + + String namespace = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_NAMESPACE); + if (!StringUtils.isEmpty(namespace)) { + mqProperties.setNamespace(namespace); + } + for (Object key : properties.keySet()) { key = StringUtils.trim(key.toString()); if (((String) key).startsWith(CanalConstants.CANAL_MQ_PROPERTIES)) { diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 3ba5404c..da62302e 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -118,3 +118,9 @@ canal.mq.acks = all # use transaction for kafka flatMessage batch produce canal.mq.transaction = true #canal.mq.properties. = +canal.mq.producerGroup = test +# Set this value to "cloud", if you want open message trace feature in aliyun. +canal.mq.accessChannel = local +# aliyun mq namespace +#canal.mq.namespace = + diff --git a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/AbstractRocektMQTest.java b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/AbstractRocektMQTest.java index 55672db1..f9952e61 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/AbstractRocektMQTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/AbstractRocektMQTest.java @@ -7,5 +7,9 @@ public abstract class AbstractRocektMQTest extends BaseCanalClientTest { public static String topic = "example"; public static String groupId = "group"; public static String nameServers = "localhost:9876"; - + public static String accessKey = ""; + public static String secretKey = ""; + public static boolean enableMessageTrace = false; + public static String accessChannel = "local"; + public static String namespace = ""; } diff --git a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java index 2a6c5c17..3c88900e 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java @@ -1,16 +1,13 @@ package com.alibaba.otter.canal.example.rocketmq; +import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; +import com.alibaba.otter.canal.protocol.Message; import java.util.List; import java.util.concurrent.TimeUnit; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.util.Assert; -import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; -import com.alibaba.otter.canal.example.kafka.AbstractKafkaTest; -import com.alibaba.otter.canal.protocol.Message; - /** * Kafka client example * @@ -34,15 +31,27 @@ public class CanalRocketMQClientExample extends AbstractRocektMQTest { } }; - public CanalRocketMQClientExample(String nameServers, String topic, String groupId){ + public CanalRocketMQClientExample(String nameServers, String topic, String groupId) { connector = new RocketMQCanalConnector(nameServers, topic, groupId, 500, false); } + public CanalRocketMQClientExample(String nameServers, String topic, String groupId, boolean enableMessageTrace, + String accessKey, String secretKey, String accessChannel, String namespace) { + connector = new RocketMQCanalConnector(nameServers, topic, groupId, accessKey, + secretKey, -1, false, enableMessageTrace, + null, accessChannel, namespace); + } + public static void main(String[] args) { try { final CanalRocketMQClientExample rocketMQClientExample = new CanalRocketMQClientExample(nameServers, topic, - groupId); + groupId, + enableMessageTrace, + accessKey, + secretKey, + accessChannel, + namespace); logger.info("## Start the rocketmq consumer: {}-{}", topic, groupId); rocketMQClientExample.start(); logger.info("## The canal rocketmq consumer is running now ......"); @@ -108,7 +117,7 @@ public class CanalRocketMQClientExample extends AbstractRocektMQTest { connector.connect(); connector.subscribe(); while (running) { - List messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS); // 获取message + List messages = connector.getListWithoutAck(1000L, TimeUnit.MILLISECONDS); // 获取message for (Message message : messages) { long batchId = message.getId(); int size = message.getEntries().size(); diff --git a/pom.xml b/pom.xml index 28a0ca4c..6d05a7ae 100644 --- a/pom.xml +++ b/pom.xml @@ -99,6 +99,7 @@ 1.8 UTF-8 3.2.18.RELEASE + 4.5.1 0.8.3 2.22.1 -server -Xms512m -Xmx1024m -Dfile.encoding=UTF-8 @@ -309,14 +310,14 @@ 3.0.2 - com.aliyun.openservices - aliware-apache-rocketmq-cloud - 1.0 + org.apache.rocketmq + rocketmq-client + ${rocketmq_version} org.apache.rocketmq - rocketmq-client - 4.3.0 + rocketmq-acl + ${rocketmq_version} javax.annotation diff --git a/server/pom.xml b/server/pom.xml index 4a7eb6ef..c1061312 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -41,6 +41,10 @@ org.apache.rocketmq rocketmq-client + + org.apache.rocketmq + rocketmq-acl + org.jboss.netty @@ -54,9 +58,5 @@ junit test - - com.aliyun.openservices - aliware-apache-rocketmq-cloud - diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index 4ee28c5e..e1987562 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -3,7 +3,7 @@ package com.alibaba.otter.canal.common; import java.util.Properties; /** - * kafka 配置项 + * MQ 配置项 * * @author machengyuan 2018-6-11 下午05:30:49 * @version 1.0.0 @@ -27,7 +27,10 @@ public class MQProperties { private String aliyunSecretKey = ""; private boolean transaction = false; // 是否开启事务 private Properties properties = new Properties(); - + private boolean enableMessageTrace = false; + private String accessChannel = null; + private String customizedTraceTopic = null; + private String namespace = ""; public static class CanalDestination { private String canalDestination; @@ -221,4 +224,62 @@ public class MQProperties { public void setProperties(Properties properties) { this.properties = properties; } + + public boolean isEnableMessageTrace() { + return enableMessageTrace; + } + + public void setEnableMessageTrace(boolean enableMessageTrace) { + this.enableMessageTrace = enableMessageTrace; + } + + public String getAccessChannel() { + return accessChannel; + } + + public void setAccessChannel(String accessChannel) { + this.accessChannel = accessChannel; + } + + public String getCustomizedTraceTopic() { + return customizedTraceTopic; + } + + public void setCustomizedTraceTopic(String customizedTraceTopic) { + this.customizedTraceTopic = customizedTraceTopic; + } + + public String getNamespace() { + return namespace; + } + + public void setNamespace(String namespace) { + this.namespace = namespace; + } + + @Override public String toString() { + return "MQProperties{" + + "servers='" + servers + '\'' + + ", retries=" + retries + + ", batchSize=" + batchSize + + ", lingerMs=" + lingerMs + + ", maxRequestSize=" + maxRequestSize + + ", bufferMemory=" + bufferMemory + + ", filterTransactionEntry=" + filterTransactionEntry + + ", producerGroup='" + producerGroup + '\'' + + ", canalBatchSize=" + canalBatchSize + + ", canalGetTimeout=" + canalGetTimeout + + ", flatMessage=" + flatMessage + + ", compressionType='" + compressionType + '\'' + + ", acks='" + acks + '\'' + + ", aliyunAccessKey='" + aliyunAccessKey + '\'' + + ", aliyunSecretKey='" + aliyunSecretKey + '\'' + + ", transaction=" + transaction + + ", properties=" + properties + + ", enableMessageTrace=" + enableMessageTrace + + ", accessChannel='" + accessChannel + '\'' + + ", customizedTraceTopic='" + customizedTraceTopic + '\'' + + ", namespace='" + namespace + '\'' + + '}'; + } } diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index aaa658e2..e16ab3d3 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -4,17 +4,20 @@ import java.util.List; import java.util.Map; import org.apache.commons.lang.StringUtils; +import org.apache.rocketmq.acl.common.AclClientRPCHook; +import org.apache.rocketmq.acl.common.SessionCredentials; +import org.apache.rocketmq.client.AccessChannel; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.MessageQueueSelector; +import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.remoting.RPCHook; import org.apache.rocketmq.remoting.exception.RemotingException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.common.CanalMessageSerializer; @@ -23,14 +26,13 @@ import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.server.exception.CanalServerException; import com.alibaba.otter.canal.spi.CanalMQProducer; -import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials; -import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook; public class CanalRocketMQProducer implements CanalMQProducer { private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class); private DefaultMQProducer defaultMQProducer; private MQProperties mqProperties; + private static final String CLOUD_ACCESS_CHANNEL = "cloud"; @Override public void init(MQProperties rocketMQProperties) { @@ -41,10 +43,16 @@ public class CanalRocketMQProducer implements CanalMQProducer { SessionCredentials sessionCredentials = new SessionCredentials(); sessionCredentials.setAccessKey(rocketMQProperties.getAliyunAccessKey()); sessionCredentials.setSecretKey(rocketMQProperties.getAliyunSecretKey()); - rpcHook = new ClientRPCHook(sessionCredentials); + rpcHook = new AclClientRPCHook(sessionCredentials); } - defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(), rpcHook); + defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(), rpcHook, mqProperties.isEnableMessageTrace(), mqProperties.getCustomizedTraceTopic()); + if (CLOUD_ACCESS_CHANNEL.equals(rocketMQProperties.getAccessChannel())){ + defaultMQProducer.setAccessChannel(AccessChannel.CLOUD); + } + if (!StringUtils.isEmpty(mqProperties.getNamespace())){ + defaultMQProducer.setNamespace(mqProperties.getNamespace()); + } defaultMQProducer.setNamesrvAddr(rocketMQProperties.getServers()); defaultMQProducer.setRetryTimesWhenSendFailed(rocketMQProperties.getRetries()); defaultMQProducer.setVipChannelEnabled(false); @@ -80,6 +88,22 @@ public class CanalRocketMQProducer implements CanalMQProducer { } } + private void sendMessage(Message message, int partition) throws Exception{ + SendResult sendResult = this.defaultMQProducer.send(message, new MessageQueueSelector() { + @Override + public MessageQueue select(List mqs, Message msg, Object arg) { + if (partition > mqs.size()) { + return mqs.get(partition % mqs.size()); + } else { + return mqs.get(partition); + } + } + }, null); + if (logger.isDebugEnabled()) { + logger.debug("Send Message Result: {}", sendResult); + } + } + public void send(final MQProperties.CanalDestination destination, String topicName, com.alibaba.otter.canal.protocol.Message data) throws Exception { if (!mqProperties.getFlatMessage()) { @@ -102,17 +126,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { Message message = new Message(topicName, CanalMessageSerializer.serializer(dataPartition, mqProperties.isFilterTransactionEntry())); - this.defaultMQProducer.send(message, new MessageQueueSelector() { - - @Override - public MessageQueue select(List mqs, Message msg, Object arg) { - if (index > mqs.size()) { - return mqs.get(index % mqs.size()); - } else { - return mqs.get(index); - } - } - }, null); + sendMessage(message, index); } catch (Exception e) { logger.error("send flat message to hashed partition error", e); throw e; @@ -129,17 +143,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { destination.getCanalDestination(), partition); } - this.defaultMQProducer.send(message, new MessageQueueSelector() { - - @Override - public MessageQueue select(List mqs, Message msg, Object arg) { - if (partition > mqs.size()) { - return mqs.get(partition % mqs.size()); - } else { - return mqs.get(partition); - } - } - }, null); + sendMessage(message, partition); } } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { logger.error("Send message error!", e); @@ -166,17 +170,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { try { Message message = new Message(topicName, JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue).getBytes()); - this.defaultMQProducer.send(message, new MessageQueueSelector() { - - @Override - public MessageQueue select(List mqs, Message msg, Object arg) { - if (index > mqs.size()) { - return mqs.get(index % mqs.size()); - } else { - return mqs.get(index); - } - } - }, null); + sendMessage(message, index); } catch (Exception e) { logger.error("send flat message to hashed partition error", e); throw e; @@ -194,17 +188,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } Message message = new Message(topicName, JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue).getBytes()); - this.defaultMQProducer.send(message, new MessageQueueSelector() { - - @Override - public MessageQueue select(List mqs, Message msg, Object arg) { - if (partition > mqs.size()) { - return mqs.get(partition % mqs.size()); - } else { - return mqs.get(partition); - } - } - }, null); + sendMessage(message, partition); } catch (Exception e) { logger.error("send flat message to fixed partition error", e); throw e; From 37fc248ebfbe7352319c2fb67c42833e48a5d96e Mon Sep 17 00:00:00 2001 From: Marsss___ <31469905+yabola@users.noreply.github.com> Date: Mon, 27 May 2019 23:54:25 +0800 Subject: [PATCH 03/17] =?UTF-8?q?=E5=88=A0=E9=99=A4=E6=93=8D=E4=BD=9C?= =?UTF-8?q?=E8=AE=A1=E7=AE=97hash=E5=80=BC=E7=9A=84=E6=83=85=E5=86=B5?= =?UTF-8?q?=E6=B2=A1=E6=9C=89=E8=80=83=E8=99=91=EF=BC=8C=E5=AF=BC=E8=87=B4?= =?UTF-8?q?=E5=88=A0=E9=99=A4=E6=97=B6=E7=9A=84=E5=88=86=E5=8C=BA=E5=92=8C?= =?UTF-8?q?=E6=8F=92=E5=85=A5=E6=97=B6=E7=9A=84=E5=88=86=E5=8C=BA=E4=B8=8D?= =?UTF-8?q?=E4=B8=80=E8=87=B4=20(#1848)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/common/MQMessageUtils.java | 21 ++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java index fc27baca..d1598809 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -230,10 +230,25 @@ public class MQMessageUtils { } } } else { - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) { - hashCode = hashCode ^ column.getValue().hashCode(); + try { + CanalEntry.EventType eventType = RowChange.parseFrom(entry.getStoreValue()).getEventType(); + if(eventType == CanalEntry.EventType.DELETE){ + for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { + if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) { + hashCode = hashCode ^ column.getValue().hashCode(); + } + } } + else { + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) { + hashCode = hashCode ^ column.getValue().hashCode(); + } + } + } + } + catch (InvalidProtocolBufferException e) { + e.printStackTrace(); } } From 4075a0623f0f0ca527dfb6e97ba31fe1039ff78d Mon Sep 17 00:00:00 2001 From: xiayuan Date: Mon, 27 May 2019 23:56:25 +0800 Subject: [PATCH 04/17] set the validationQuery while testWhileIdle is true (#1827) --- deployer/src/main/resources/spring/tsdb/h2-tsdb.xml | 1 + 1 file changed, 1 insertion(+) diff --git a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml index c05b5200..6c34241d 100644 --- a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml +++ b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml @@ -43,6 +43,7 @@ + From f9989e085674c0f5068f9568448329fa8513d16e Mon Sep 17 00:00:00 2001 From: nieyuan1991 Date: Tue, 28 May 2019 00:02:25 +0800 Subject: [PATCH 05/17] =?UTF-8?q?fix:=E6=9C=AA=E6=8D=95=E8=8E=B7kafka?= =?UTF-8?q?=E4=BA=8B=E5=8A=A1=E5=BC=82=E5=B8=B8=EF=BC=8C=E9=80=A0=E6=88=90?= =?UTF-8?q?canal=20ack=E5=BC=82=E5=B8=B8=20(#1814)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/CanalKafkaProducer.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 89690a29..51fce634 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -102,10 +102,10 @@ public class CanalKafkaProducer implements CanalMQProducer { producerTmp = producer2; } - if (kafkaProperties.getTransaction()) { - producerTmp.beginTransaction(); - } try { + if (kafkaProperties.getTransaction()) { + producerTmp.beginTransaction(); + } if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) { // 动态topic Map messageMap = MQMessageUtils.messageTopics(message, @@ -130,7 +130,11 @@ public class CanalKafkaProducer implements CanalMQProducer { } catch (Throwable e) { logger.error(e.getMessage(), e); if (kafkaProperties.getTransaction()) { - producerTmp.abortTransaction(); + try { + producerTmp.abortTransaction(); + } catch (Exception e1) { + logger.error(e1.getMessage(), e1); + } } callback.rollback(); } From 8df6d6fed8b9256a4591d5e0499d36a9bef3cf33 Mon Sep 17 00:00:00 2001 From: soso Date: Tue, 28 May 2019 00:03:28 +0800 Subject: [PATCH 06/17] =?UTF-8?q?Fix:=20=E4=B8=BB=E5=A4=87=E5=88=87?= =?UTF-8?q?=E6=8D=A2=E5=90=8E=E6=B2=A1=E6=9C=89=E5=BE=80Kafka=E5=8F=91?= =?UTF-8?q?=E9=80=81=E6=95=B0=E6=8D=AE=20(#1810)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 发生主备切换时，调用canalMQStarter启停对应的destination。 --- .../otter/canal/deployer/CanalController.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index d8ca17fd..c0692227 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -152,6 +152,9 @@ public class CanalController { try { MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); embededCanalServer.start(destination); + if (canalMQStarter != null) { + canalMQStarter.startDestination(destination); + } } finally { MDC.remove(CanalConstants.MDC_DESTINATION); } @@ -160,6 +163,9 @@ public class CanalController { public void processActiveExit() { try { MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); + if (canalMQStarter != null) { + canalMQStarter.stopDestination(destination); + } embededCanalServer.stop(destination); } finally { MDC.remove(CanalConstants.MDC_DESTINATION); @@ -234,9 +240,6 @@ public class CanalController { ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination); if (!config.getLazy() && !runningMonitor.isStart()) { runningMonitor.start(); - if (canalMQStarter != null) { - canalMQStarter.startDestination(destination); - } } } } @@ -245,9 +248,6 @@ public class CanalController { // 此处的stop，代表强制退出，非HA机制，所以需要退出HA的monitor和配置信息 InstanceConfig config = instanceConfigs.remove(destination); if (config != null) { - if (canalMQStarter != null) { - canalMQStarter.stopDestination(destination); - } embededCanalServer.stop(destination); ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination); if (runningMonitor.isStart()) { From ce7948d6e72f2399caeeb279e1702225c89cdad3 Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 28 May 2019 00:03:55 +0800 Subject: [PATCH 07/17] =?UTF-8?q?fix=20Message=E8=BD=ACDml=E5=AF=B9null?= =?UTF-8?q?=E5=80=BC=E7=9A=84bug=20(#1808)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Message转Dml对null值的判断 * Message转Dml对null值的判断 * Message转Dml对null值的判断 --- .../client/adapter/support/MessageUtil.java | 34 ++++++++++++------- 1 file changed, 22 insertions(+), 12 deletions(-) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java index 1e24d9ad..618f81d6 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java @@ -76,11 +76,16 @@ public class MessageUtil { dml.getPkNames().add(column.getName()); } } - row.put(column.getName(), - JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(), - column.getValue(), - column.getSqlType(), - column.getMysqlType())); + if (column.getIsNull()) { + row.put(column.getName(), null); + } else { + row.put(column.getName(), + JdbcTypeUtil.typeConvert(dml.getTable(), + column.getName(), + column.getValue(), + column.getSqlType(), + column.getMysqlType())); + } // 获取update为true的字段 if (column.getUpdated()) { updateSet.add(column.getName()); @@ -94,11 +99,16 @@ public class MessageUtil { Map rowOld = new LinkedHashMap<>(); for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { if (updateSet.contains(column.getName())) { - rowOld.put(column.getName(), - JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(), - column.getValue(), - column.getSqlType(), - column.getMysqlType())); + if (column.getIsNull()) { + rowOld.put(column.getName(), null); + } else { + rowOld.put(column.getName(), + JdbcTypeUtil.typeConvert(dml.getTable(), + column.getName(), + column.getValue(), + column.getSqlType(), + column.getMysqlType())); + } } } // update操作将记录修改前的值 @@ -162,8 +172,8 @@ public class MessageUtil { return dml; } - private static List> changeRows(String table, List> rows, Map sqlTypes, - Map mysqlTypes) { + private static List> changeRows(String table, List> rows, + Map sqlTypes, Map mysqlTypes) { List> result = new ArrayList<>(); for (Map row : rows) { Map resultRow = new LinkedHashMap<>(); From e808c6ec70d5be731d0e8ba18f9eed10a29336c3 Mon Sep 17 00:00:00 2001 From: lulu2panpan Date: Tue, 28 May 2019 00:05:13 +0800 Subject: [PATCH 08/17] =?UTF-8?q?AbstractCanalInstance=E7=9A=84subscribeCh?= =?UTF-8?q?ange=E6=96=B9=E6=B3=95=EF=BC=8C=E5=A2=9E=E5=8A=A0=E5=AF=B9Abstr?= =?UTF-8?q?actEventParser=E7=9A=84=E7=B1=BB=E5=9E=8B=E5=88=A4=E6=96=AD?= =?UTF-8?q?=EF=BC=9BMysqlEventParser=E5=AF=B9Gtid=E7=9A=84=E5=A4=84?= =?UTF-8?q?=E7=90=86=E9=9C=80=E8=A6=81=E8=80=83=E8=99=91=E5=85=BC=E5=AE=B9?= =?UTF-8?q?=E6=80=A7=20(#1824)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * AbstractCanalInstance的subscribeChange方法，增加对AbstractEventParser的类型判断，以支持用户的自定义CanalEventParser * MysqlEventParser对Gtid模式的处理需要考虑兼容性，支持系统平滑的从非gtid模式升级为gtid模式 --- .../canal/instance/core/AbstractCanalInstance.java | 8 ++++++-- .../canal/parse/inbound/AbstractEventParser.java | 4 ++-- .../canal/parse/inbound/mysql/MysqlEventParser.java | 13 ++++++++----- 3 files changed, 16 insertions(+), 9 deletions(-) diff --git a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java index b1b64d5b..6993a55a 100644 --- a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java +++ b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java @@ -53,10 +53,14 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can // 处理group的模式 List eventParsers = ((GroupEventParser) eventParser).getEventParsers(); for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动 - ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter); + if(singleEventParser instanceof AbstractEventParser) { + ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter); + } } } else { - ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter); + if(eventParser instanceof AbstractEventParser) { + ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter); + } } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index 73b35df6..409c15d2 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -242,7 +242,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle if (parallel) { // build stage processor multiStageCoprocessor = buildMultiStageCoprocessor(); - if (isGTIDMode()) { + if (isGTIDMode() && StringUtils.isNotEmpty(startPosition.getGtid())) { // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据 GTIDSet gtidSet = MysqlGTIDSet.parse(startPosition.getGtid()); ((MysqlMultiStageCoprocessor) multiStageCoprocessor).setGtidSet(gtidSet); @@ -260,7 +260,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle } } } else { - if (isGTIDMode()) { + if (isGTIDMode() && StringUtils.isNotEmpty(startPosition.getGtid())) { // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据 erosaConnection.dump(MysqlGTIDSet.parse(startPosition.getGtid()), sinkHandler); } else { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index d2ce8356..68eb5062 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -355,11 +355,14 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE // GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instanc配置中的 LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination); if (logPosition != null) { - return logPosition.getPostion(); - } - - if (masterPosition != null && StringUtils.isNotEmpty(masterPosition.getGtid())) { - return masterPosition; + // 如果以前是非GTID模式，后来调整为了GTID模式，那么为了保持兼容，需要判断gtid是否为空 + if (StringUtils.isNotEmpty(logPosition.getPostion().getGtid())) { + return logPosition.getPostion(); + } + }else { + if (masterPosition != null && StringUtils.isNotEmpty(masterPosition.getGtid())) { + return masterPosition; + } } } From 058e1fc546d4d2a7bd50c2a2e748c088b2cb5258 Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 28 May 2019 00:14:20 +0800 Subject: [PATCH 09/17] fixed delete message send mq bugfix --- .../otter/canal/common/MQMessageUtils.java | 31 +++++++------------ 1 file changed, 12 insertions(+), 19 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java index d1598809..9161f8a5 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -222,33 +222,26 @@ public class MQMessageUtils { } else { for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { int hashCode = database.hashCode(); + CanalEntry.EventType eventType = rowChange.getEventType(); + List columns = null; + if (eventType == CanalEntry.EventType.DELETE) { + columns = rowData.getBeforeColumnsList(); + } else { + columns = rowData.getAfterColumnsList(); + } + if (hashMode.autoPkHash) { // isEmpty use default pkNames - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + for (CanalEntry.Column column : columns) { if (column.getIsKey()) { hashCode = hashCode ^ column.getValue().hashCode(); } } } else { - try { - CanalEntry.EventType eventType = RowChange.parseFrom(entry.getStoreValue()).getEventType(); - if(eventType == CanalEntry.EventType.DELETE){ - for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { - if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) { - hashCode = hashCode ^ column.getValue().hashCode(); - } - } + for (CanalEntry.Column column : columns) { + if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) { + hashCode = hashCode ^ column.getValue().hashCode(); } - else { - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) { - hashCode = hashCode ^ column.getValue().hashCode(); - } - } - } - } - catch (InvalidProtocolBufferException e) { - e.printStackTrace(); } } From 95602f4784f809bf2ed39ac17b562597304084ae Mon Sep 17 00:00:00 2001 From: lanxinxu <48084525+lanxinxu@users.noreply.github.com> Date: Tue, 18 Jun 2019 14:57:59 +0800 Subject: [PATCH 10/17] =?UTF-8?q?=E5=B9=B6=E8=A1=8C=E8=A7=A3=E6=9E=90?= =?UTF-8?q?=E6=A8=A1=E5=BC=8F=E4=B8=8B=E6=9C=89=E4=B8=A2=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E9=A3=8E=E9=99=A9=20(#1857)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 70aba4f1..5c5ad87a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -441,6 +441,8 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement @Override public void handleEventException(final Throwable ex, final long sequence, final Object event) { + //异常上抛，否则processEvents的逻辑会默认会mark为成功执行，有丢数据风险 + throw new CanalParseException(ex); } @Override From 75b4911caf188e2318fbade94d23d2e2f3b32ad6 Mon Sep 17 00:00:00 2001 From: lanxinxu <48084525+lanxinxu@users.noreply.github.com> Date: Tue, 18 Jun 2019 14:59:30 +0800 Subject: [PATCH 11/17] =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E5=AE=89=E5=85=A8?= =?UTF-8?q?=E7=9A=84=E8=8E=B7=E5=8F=96meta=20(#1866)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/parse/inbound/mysql/dbsync/TableMetaCache.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java index 25554c29..8c9d0e39 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java @@ -81,7 +81,7 @@ public class TableMetaCache { } } - private TableMeta getTableMetaByDB(String fullname) throws IOException { + private synchronized TableMeta getTableMetaByDB(String fullname) throws IOException { try { ResultSetPacket packet = connection.query("show create table " + fullname); String[] names = StringUtils.split(fullname, "`.`"); @@ -159,7 +159,7 @@ public class TableMetaCache { return getTableMeta(schema, table, true, position); } - public TableMeta getTableMeta(String schema, String table, boolean useCache, EntryPosition position) { + public synchronized TableMeta getTableMeta(String schema, String table, boolean useCache, EntryPosition position) { TableMeta tableMeta = null; if (tableMetaTSDB != null) { tableMeta = tableMetaTSDB.find(schema, table); From 5ac0f7b710c02d1612ca2c5775e6d572bace5512 Mon Sep 17 00:00:00 2001 From: Z6N Date: Tue, 18 Jun 2019 15:01:00 +0800 Subject: [PATCH 12/17] =?UTF-8?q?=E6=8C=89=E4=B8=BB=E9=94=AE=E6=95=A3?= =?UTF-8?q?=E5=88=97=EF=BC=8C=E6=9C=AA=E8=AE=BE=E7=BD=AE=E4=B8=BB=E9=94=AE?= =?UTF-8?q?=E6=97=B6=E6=97=A0=E9=99=90=E7=A9=BA=E6=8C=87=E9=92=88=E5=BC=82?= =?UTF-8?q?=E5=B8=B8=20(#1889)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 当按注解散列时，如果监控的表未显示设置主键，将会导致flatMessage.getPkNames()为空，导致无限空指针异常 --- .../alibaba/otter/canal/common/MQMessageUtils.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java index 9161f8a5..de06d3b2 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -446,12 +446,14 @@ public class MQMessageUtils { int idx = 0; for (Map row : flatMessage.getData()) { int hashCode = database.hashCode(); - for (String pkName : pkNames) { - String value = row.get(pkName); - if (value == null) { - value = ""; + if (pkNames != null) { + for (String pkName : pkNames) { + String value = row.get(pkName); + if (value == null) { + value = ""; + } + hashCode = hashCode ^ value.hashCode(); } - hashCode = hashCode ^ value.hashCode(); } int pkHash = Math.abs(hashCode) % partitionsNum; From 74489d1bba073abd343c042b2733d201d9dd8895 Mon Sep 17 00:00:00 2001 From: BillowX <762348474@qq.com> Date: Tue, 18 Jun 2019 15:02:14 +0800 Subject: [PATCH 13/17] =?UTF-8?q?=E5=BD=93MQ=E9=85=8D=E7=BD=AE=E4=B8=BAKaf?= =?UTF-8?q?ka=E6=97=B6=EF=BC=8C=E6=94=AF=E6=8C=81=E9=85=8D=E7=BD=AEKerbero?= =?UTF-8?q?s=E8=AE=A4=E8=AF=81=20(#1895)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/deployer/CanalConstants.java | 87 ++++++++++--------- .../otter/canal/deployer/CanalStater.java | 34 +++++--- deployer/src/main/resources/canal.properties | 7 ++ .../otter/canal/common/MQProperties.java | 31 +++++++ .../otter/canal/kafka/CanalKafkaProducer.java | 48 +++++++--- 5 files changed, 140 insertions(+), 67 deletions(-) diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java index 78f22e72..6d0d9794 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java @@ -10,55 +10,56 @@ import java.text.MessageFormat; */ public class CanalConstants { - public static final String MDC_DESTINATION = "destination"; - public static final String ROOT = "canal"; - public static final String CANAL_ID = ROOT + "." + "id"; - public static final String CANAL_IP = ROOT + "." + "ip"; - public static final String CANAL_PORT = ROOT + "." + "port"; - public static final String CANAL_METRICS_PULL_PORT = ROOT + "." + "metrics.pull.port"; - public static final String CANAL_ZKSERVERS = ROOT + "." + "zkServers"; - public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty"; + public static final String MDC_DESTINATION = "destination"; + public static final String ROOT = "canal"; + public static final String CANAL_ID = ROOT + "." + "id"; + public static final String CANAL_IP = ROOT + "." + "ip"; + public static final String CANAL_PORT = ROOT + "." + "port"; + public static final String CANAL_METRICS_PULL_PORT = ROOT + "." + "metrics.pull.port"; + public static final String CANAL_ZKSERVERS = ROOT + "." + "zkServers"; + public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty"; - public static final String CANAL_DESTINATIONS = ROOT + "." + "destinations"; - public static final String CANAL_AUTO_SCAN = ROOT + "." + "auto.scan"; - public static final String CANAL_AUTO_SCAN_INTERVAL = ROOT + "." + "auto.scan.interval"; - public static final String CANAL_CONF_DIR = ROOT + "." + "conf.dir"; - public static final String CANAL_SERVER_MODE = ROOT + "." + "serverMode"; + public static final String CANAL_DESTINATIONS = ROOT + "." + "destinations"; + public static final String CANAL_AUTO_SCAN = ROOT + "." + "auto.scan"; + public static final String CANAL_AUTO_SCAN_INTERVAL = ROOT + "." + "auto.scan.interval"; + public static final String CANAL_CONF_DIR = ROOT + "." + "conf.dir"; + public static final String CANAL_SERVER_MODE = ROOT + "." + "serverMode"; - public static final String CANAL_DESTINATION_SPLIT = ","; - public static final String GLOBAL_NAME = "global"; + public static final String CANAL_DESTINATION_SPLIT = ","; + public static final String GLOBAL_NAME = "global"; - public static final String INSTANCE_MODE_TEMPLATE = ROOT + "." + "instance.{0}.mode"; - public static final String INSTANCE_LAZY_TEMPLATE = ROOT + "." + "instance.{0}.lazy"; - public static final String INSTANCE_MANAGER_ADDRESS_TEMPLATE = ROOT + "." + "instance.{0}.manager.address"; - public static final String INSTANCE_SPRING_XML_TEMPLATE = ROOT + "." + "instance.{0}.spring.xml"; + public static final String INSTANCE_MODE_TEMPLATE = ROOT + "." + "instance.{0}.mode"; + public static final String INSTANCE_LAZY_TEMPLATE = ROOT + "." + "instance.{0}.lazy"; + public static final String INSTANCE_MANAGER_ADDRESS_TEMPLATE = ROOT + "." + "instance.{0}.manager.address"; + public static final String INSTANCE_SPRING_XML_TEMPLATE = ROOT + "." + "instance.{0}.spring.xml"; - public static final String CANAL_DESTINATION_PROPERTY = ROOT + ".instance.destination"; - - public static final String CANAL_SOCKETCHANNEL = ROOT + "." + "socketChannel"; - - public static final String CANAL_MQ_SERVERS = ROOT + "." + "mq.servers"; - public static final String CANAL_MQ_RETRIES = ROOT + "." + "mq.retries"; - public static final String CANAL_MQ_BATCHSIZE = ROOT + "." + "mq.batchSize"; - public static final String CANAL_MQ_LINGERMS = ROOT + "." + "mq.lingerMs"; - public static final String CANAL_MQ_MAXREQUESTSIZE = ROOT + "." + "mq.maxRequestSize"; - public static final String CANAL_MQ_BUFFERMEMORY = ROOT + "." + "mq.bufferMemory"; - public static final String CANAL_MQ_CANALBATCHSIZE = ROOT + "." + "mq.canalBatchSize"; - public static final String CANAL_MQ_CANALGETTIMEOUT = ROOT + "." + "mq.canalGetTimeout"; - public static final String CANAL_MQ_FLATMESSAGE = ROOT + "." + "mq.flatMessage"; - public static final String CANAL_MQ_COMPRESSION_TYPE = ROOT + "." + "mq.compressionType"; - public static final String CANAL_MQ_ACKS = ROOT + "." + "mq.acks"; - public static final String CANAL_MQ_TRANSACTION = ROOT + "." + "mq.transaction"; - public static final String CANAL_MQ_PRODUCERGROUP = ROOT + "." + "mq.producerGroup"; - public static final String CANAL_ALIYUN_ACCESSKEY = ROOT + "." + "aliyun.accessKey"; - public static final String CANAL_ALIYUN_SECRETKEY = ROOT + "." + "aliyun.secretKey"; - public static final String CANAL_MQ_PROPERTIES = ROOT + "." + "mq.properties"; - public static final String CANAL_MQ_ENABLE_MESSAGE_TRACE = ROOT + "." + "mq.enableMessageTrace"; - public static final String CANAL_MQ_ACCESS_CHANNEL = ROOT + "." + "mq.accessChannel"; - public static final String CANAL_MQ_CUSTOMIZED_TRACE_TOPIC = ROOT + "." + "mq.customizedTraceTopic"; - public static final String CANAL_MQ_NAMESPACE = ROOT + "." + "mq.namespace"; + public static final String CANAL_DESTINATION_PROPERTY = ROOT + ".instance.destination"; + public static final String CANAL_SOCKETCHANNEL = ROOT + "." + "socketChannel"; + public static final String CANAL_MQ_SERVERS = ROOT + "." + "mq.servers"; + public static final String CANAL_MQ_RETRIES = ROOT + "." + "mq.retries"; + public static final String CANAL_MQ_BATCHSIZE = ROOT + "." + "mq.batchSize"; + public static final String CANAL_MQ_LINGERMS = ROOT + "." + "mq.lingerMs"; + public static final String CANAL_MQ_MAXREQUESTSIZE = ROOT + "." + "mq.maxRequestSize"; + public static final String CANAL_MQ_BUFFERMEMORY = ROOT + "." + "mq.bufferMemory"; + public static final String CANAL_MQ_CANALBATCHSIZE = ROOT + "." + "mq.canalBatchSize"; + public static final String CANAL_MQ_CANALGETTIMEOUT = ROOT + "." + "mq.canalGetTimeout"; + public static final String CANAL_MQ_FLATMESSAGE = ROOT + "." + "mq.flatMessage"; + public static final String CANAL_MQ_COMPRESSION_TYPE = ROOT + "." + "mq.compressionType"; + public static final String CANAL_MQ_ACKS = ROOT + "." + "mq.acks"; + public static final String CANAL_MQ_TRANSACTION = ROOT + "." + "mq.transaction"; + public static final String CANAL_MQ_PRODUCERGROUP = ROOT + "." + "mq.producerGroup"; + public static final String CANAL_ALIYUN_ACCESSKEY = ROOT + "." + "aliyun.accessKey"; + public static final String CANAL_ALIYUN_SECRETKEY = ROOT + "." + "aliyun.secretKey"; + public static final String CANAL_MQ_PROPERTIES = ROOT + "." + "mq.properties"; + public static final String CANAL_MQ_ENABLE_MESSAGE_TRACE = ROOT + "." + "mq.enableMessageTrace"; + public static final String CANAL_MQ_ACCESS_CHANNEL = ROOT + "." + "mq.accessChannel"; + public static final String CANAL_MQ_CUSTOMIZED_TRACE_TOPIC = ROOT + "." + "mq.customizedTraceTopic"; + public static final String CANAL_MQ_NAMESPACE = ROOT + "." + "mq.namespace"; + public static final String CANAL_MQ_KAFKA_KERBEROS_ENABLE = ROOT + "." + "mq.kafka.kerberos.enable"; + public static final String CANAL_MQ_KAFKA_KERBEROS_KRB5FILEPATH = ROOT + "." + "mq.kafka.kerberos.krb5FilePath"; + public static final String CANAL_MQ_KAFKA_KERBEROS_JAASFILEPATH = ROOT + "." + "mq.kafka.kerberos.jaasFilePath"; public static String getInstanceModeKey(String destination) { return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java index 41595461..09a227be 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java @@ -1,15 +1,5 @@ package com.alibaba.otter.canal.deployer; -import java.io.File; -import java.io.FileFilter; -import java.util.Arrays; -import java.util.List; -import java.util.Properties; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.kafka.CanalKafkaProducer; import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer; @@ -18,6 +8,15 @@ import com.alibaba.otter.canal.spi.CanalMQProducer; import com.google.common.base.Function; import com.google.common.base.Joiner; import com.google.common.collect.Lists; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileFilter; +import java.util.Arrays; +import java.util.List; +import java.util.Properties; /** * Canal server 启动类 @@ -227,6 +226,21 @@ public class CanalStater { mqProperties.setNamespace(namespace); } + String kafkaKerberosEnable = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_KAFKA_KERBEROS_ENABLE); + if (!StringUtils.isEmpty(kafkaKerberosEnable)) { + mqProperties.setKerberosEnable(Boolean.valueOf(kafkaKerberosEnable)); + } + + String kafkaKerberosKrb5Filepath = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_KAFKA_KERBEROS_KRB5FILEPATH); + if (!StringUtils.isEmpty(kafkaKerberosKrb5Filepath)) { + mqProperties.setKerberosKrb5FilePath(kafkaKerberosKrb5Filepath); + } + + String kafkaKerberosJaasFilepath = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_KAFKA_KERBEROS_JAASFILEPATH); + if (!StringUtils.isEmpty(kafkaKerberosJaasFilepath)) { + mqProperties.setKerberosJaasFilePath(kafkaKerberosJaasFilepath); + } + for (Object key : properties.keySet()) { key = StringUtils.trim(key.toString()); if (((String) key).startsWith(CanalConstants.CANAL_MQ_PROPERTIES)) { diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index da62302e..d7f6b102 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -124,3 +124,10 @@ canal.mq.accessChannel = local # aliyun mq namespace #canal.mq.namespace = +################################################## +######### Kafka Kerberos Info ############# +################################################## +canal.mq.kafka.kerberos.enable = false +canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf" +canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf" + diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index e1987562..41e41684 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -31,6 +31,10 @@ public class MQProperties { private String accessChannel = null; private String customizedTraceTopic = null; private String namespace = ""; + private boolean kerberosEnable = false; //kafka集群是否启动Kerberos认证 + private String kerberosKrb5FilePath = ""; //启动Kerberos认证时配置为krb5.conf文件的路径 + private String kerberosJaasFilePath = ""; //启动Kerberos认证时配置为jaas.conf文件的路径 + public static class CanalDestination { private String canalDestination; @@ -257,6 +261,30 @@ public class MQProperties { this.namespace = namespace; } + public boolean isKerberosEnable() { + return kerberosEnable; + } + + public void setKerberosEnable(boolean kerberosEnable) { + this.kerberosEnable = kerberosEnable; + } + + public String getKerberosKrb5FilePath() { + return kerberosKrb5FilePath; + } + + public void setKerberosKrb5FilePath(String kerberosKrb5FilePath) { + this.kerberosKrb5FilePath = kerberosKrb5FilePath; + } + + public String getKerberosJaasFilePath() { + return kerberosJaasFilePath; + } + + public void setKerberosJaasFilePath(String kerberosJaasFilePath) { + this.kerberosJaasFilePath = kerberosJaasFilePath; + } + @Override public String toString() { return "MQProperties{" + "servers='" + servers + '\'' + @@ -280,6 +308,9 @@ public class MQProperties { ", accessChannel='" + accessChannel + '\'' + ", customizedTraceTopic='" + customizedTraceTopic + '\'' + ", namespace='" + namespace + '\'' + + ", kerberosEnable='" + kerberosEnable + '\'' + + ", kerberosKrb5FilePath='" + kerberosKrb5FilePath + '\'' + + ", kerberosJaasFilePath='" + kerberosJaasFilePath + '\'' + '}'; } } diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 51fce634..99f2010a 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -1,19 +1,5 @@ package com.alibaba.otter.canal.kafka; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.concurrent.ExecutionException; - -import org.apache.commons.lang.StringUtils; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.common.MQMessageUtils; @@ -21,6 +7,20 @@ import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.spi.CanalMQProducer; +import org.apache.commons.lang.StringUtils; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.ExecutionException; /** * kafka producer 主操作类 @@ -59,6 +59,26 @@ public class CanalKafkaProducer implements CanalMQProducer { } else { properties.put("retries", kafkaProperties.getRetries()); } + + if (kafkaProperties.isKerberosEnable()){ + File krb5File = new File(kafkaProperties.getKerberosKrb5FilePath()); + File jaasFile = new File(kafkaProperties.getKerberosJaasFilePath()); + if(krb5File.exists() && jaasFile.exists()){ + //配置kerberos认证，需要使用绝对路径 + System.setProperty("java.security.krb5.conf", + krb5File.getAbsolutePath()); + System.setProperty("java.security.auth.login.config", + jaasFile.getAbsolutePath()); + System.setProperty("javax.security.auth.useSubjectCredsOnly", "false"); + properties.put("security.protocol", "SASL_PLAINTEXT"); + properties.put("sasl.kerberos.service.name", "kafka"); + }else{ + String errorMsg = "ERROR # The kafka kerberos configuration file does not exist! please check it"; + logger.error(errorMsg); + throw new RuntimeException(errorMsg); + } + } + if (!kafkaProperties.getFlatMessage()) { properties.put("value.serializer", MessageSerializer.class.getName()); producer = new KafkaProducer(properties); From 6c494f8892a1c0e54fe51702f677661dfe3b7afb Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 18 Jun 2019 21:28:55 +0800 Subject: [PATCH 14/17] add rocketmq-client pom version --- client-adapter/launcher/pom.xml | 1 + 1 file changed, 1 insertion(+) diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index 27582d5f..e93956c9 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -62,6 +62,7 @@ org.apache.rocketmq rocketmq-client + 4.5.1 From 5bea876ec98c3f4e3ca0a2eb21996f8ee6cf9bad Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 18 Jun 2019 21:44:52 +0800 Subject: [PATCH 15/17] fixed issue #1352, add pkNames for flatMessage partition --- .../main/java/com/alibaba/otter/canal/common/MQMessageUtils.java | 1 + 1 file changed, 1 insertion(+) diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java index de06d3b2..0073c186 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -473,6 +473,7 @@ public class MQMessageUtils { flatMessageTmp.setMysqlType(flatMessage.getMysqlType()); flatMessageTmp.setEs(flatMessage.getEs()); flatMessageTmp.setTs(flatMessage.getTs()); + flatMessageTmp.setPkNames(flatMessage.getPkNames()); } List> data = flatMessageTmp.getData(); if (data == null) { From 8728beb4b6820d794dfb4bd1d26980f4d5541fe6 Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 18 Jun 2019 22:18:30 +0800 Subject: [PATCH 16/17] fixed compile --- .../adapter/rdb/service/RdbSyncService.java | 2 +- .../manager/CanalInstanceWithManager.java | 19 ++++++++++++++----- pom.xml | 2 +- 3 files changed, 16 insertions(+), 7 deletions(-) diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index a34bb784..350b29e9 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -272,7 +272,7 @@ public class RdbSyncService { batchExecutor.execute(insertSql.toString(), values); } catch (SQLException e) { if (skipDupException - && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001: 违反唯一约束条件"))) { + && (e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001:"))) { // ignore // TODO 增加更多关系数据库的主键冲突的错误码 } else { diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index 43eaaf33..49030fc1 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.instance.manager; import java.io.File; +import java.io.FilenameFilter; import java.net.InetSocketAddress; import java.net.URL; import java.net.URLClassLoader; @@ -117,7 +118,13 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { } else { try { File externalLibDir = new File(alarmHandlerPluginDir); - File[] jarFiles = externalLibDir.listFiles((dir1, name) -> name.endsWith(".jar")); + File[] jarFiles = externalLibDir.listFiles(new FilenameFilter() { + + @Override + public boolean accept(File dir, String name) { + return name.endsWith(".jar"); + } + }); if (jarFiles == null || jarFiles.length == 0) { throw new IllegalStateException(String.format("alarmHandlerPluginDir [%s] can't find any name endswith \".jar\" file.", alarmHandlerPluginDir)); @@ -126,14 +133,16 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { for (int i = 0; i < jarFiles.length; i++) { urls[i] = jarFiles[i].toURI().toURL(); } - ClassLoader currentClassLoader = new URLClassLoader(urls, CanalInstanceWithManager.class.getClassLoader()); - Class _alarmClass = - (Class)currentClassLoader.loadClass(alarmHandlerClass); + ClassLoader currentClassLoader = new URLClassLoader(urls, + CanalInstanceWithManager.class.getClassLoader()); + Class _alarmClass = (Class) currentClassLoader.loadClass(alarmHandlerClass); alarmHandler = _alarmClass.newInstance(); logger.info("init [{}] alarm handler success.", alarmHandlerClass); } catch (Throwable e) { String errorMsg = String.format("init alarmHandlerPluginDir [%s] alarm handler [%s] error: %s", - alarmHandlerPluginDir, alarmHandlerClass, ExceptionUtils.getFullStackTrace(e)); + alarmHandlerPluginDir, + alarmHandlerClass, + ExceptionUtils.getFullStackTrace(e)); logger.error(errorMsg); throw new CanalException(errorMsg, e); } diff --git a/pom.xml b/pom.xml index 6d05a7ae..5f875209 100644 --- a/pom.xml +++ b/pom.xml @@ -253,7 +253,7 @@ com.alibaba.fastsql fastsql - 2.0.0_preview_855 + 2.0.0_preview_914 com.alibaba From adccc0e9120bc561f5ccc6084b0ea1fa73e2693f Mon Sep 17 00:00:00 2001 From: agapple Date: Wed, 19 Jun 2019 15:04:22 +0800 Subject: [PATCH 17/17] add logger info for query_log_event --- .../com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java | 6 ++---- deployer/src/main/resources/canal.properties | 3 +-- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java index c732da99..8cc49795 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java @@ -706,10 +706,8 @@ public class QueryLogEvent extends LogEvent { * That's why you must write status vars in growing * order of code */ - if (logger.isDebugEnabled()) { - logger.debug("Query_log_event has unknown status vars (first has code: " + code - + "), skipping the rest of them"); - } + logger.error("Query_log_event has unknown status vars (first has code: " + code + + "), skipping the rest of them"); break; // Break loop } } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index d7f6b102..106005f1 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -129,5 +129,4 @@ canal.mq.accessChannel = local ################################################## canal.mq.kafka.kerberos.enable = false canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf" -canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf" - +canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf" \ No newline at end of file